Gowtham Potureddi

Posted on Jul 21

Mage AI for Data Engineering: Notebook-Style Blocks, Streaming & Data Integrations

#python #sql #interview #dataengineering

mage ai is the notebook-style orchestrator that finally collapses the classic three-tool orchestration stack — a scheduler (Airflow), an interactive IDE (Jupyter), and an integration platform (Airbyte / Fivetran) — into a single browser-native workspace where every pipeline is composed from typed blocks that preview their output live while you edit. Every operational data pipeline your team ships in 2026 — a nightly Postgres → Snowflake load, a Kafka clickstream sessionisation, a Salesforce → dbt → dashboard reverse-ETL, a Stripe webhook fan-out — has to be fast to iterate on, safe to schedule, and portable across batch and streaming without forcing the author to jump between four tools and three languages. The engineering trade-off does not live in "should we adopt Mage" — the notebook-orchestrator category has proven itself — but in which workloads mage ai wins for and how it composes with the Airflow, Dagster, Prefect, and dbt investments most teams already own.

This guide is the senior-DE walkthrough you wanted the first time an interviewer asked "walk me through the Mage block DAG model," or "compare mage ai vs dagster on lineage and dev-loop," or "design a mage streaming pipeline for a 50k events/sec Kafka topic," or "how does mage data pipeline handoff to dbt without duplicating transforms." It walks the block-based DAG (@data_loader, @transformer, @data_exporter, @sensor, @custom / scratchpad), the interactive Jupyter-like dev-loop with instant materialisation, the Kafka source + windowed transform + sink streaming architecture with offset-commit-after-sink checkpointing, the 90+ pre-built connectors and the mage tutorial dbt handoff pattern, the Mage Pro managed tier and the 2024 Airbyte acquisition roadmap convergence, and the four-tool decision matrix that separates the workloads where Mage wins (fast dev-loop, mixed batch + streaming, small-team ETL) from the workloads where Dagster / Prefect / Airflow still win (SDA lineage, dynamic Python flows, legacy DAG scale). Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the SQL practice library →, rehearse system design on the design practice library →, and sharpen the streaming axis with the streaming practice library →.

On this page

Why Mage matters in 2026
Block-based pipelines + DAG mental model
Streaming pipelines in Mage
Integrations + destinations
Mage vs Dagster / Prefect / Airflow + interview signals
Cheat sheet — Mage recipes
Frequently asked questions
Practice on PipeCode

1. Why Mage matters in 2026

One notebook IDE, one block DAG, one runtime — the orchestrator that made ETL fun to write again

The one-sentence invariant: mage ai is a notebook-style orchestrator where every pipeline is composed of typed blocks — @data_loader, @transformer, @data_exporter, @sensor, @custom — each authored in a browser IDE with live output preview, wired into a DAG by explicit upstream / downstream declarations, versioned as first-class files in Git, executed by a single unified runtime for both batch and streaming, and (as of the 2024 Airbyte acquisition) roadmapped to converge with the Airbyte connector catalogue — and the pattern you pick in 2026 becomes the pattern your platform team defends against every incoming "should we just use Dagster / Prefect / Airflow instead" review for the next three years. The mental model is not "another Airflow with a prettier UI" — it is a re-imagining of the author-run-observe loop where the primary artifact is the block, not the DAG file.

The four axes interviewers actually probe.

Dev-loop speed. Mage's live-preview loop — write a @transformer block, click Run, see the output DataFrame render below the cell — is the fastest dev-loop in the orchestrator category. Airflow demands airflow tasks test from a shell; Dagster requires materialisation via the UI; Prefect wants a Python subprocess. Mage collapses "author → preview → wire into DAG" into one browser tab. Interviewers open here because it separates people who have actually built pipelines in Mage from those who've only read the docs.
Block reusability and DAG semantics. A Mage block is a file with a decorator (@data_loader, @transformer, etc.), an input signature (the outputs of its declared upstream blocks), and a return value (a DataFrame, a dict, or nothing). Block dependencies form the DAG — no separate DAG definition, no jinja templating, no boilerplate. Interviewers probe this because it's the mechanical answer to "what makes Mage different from Airflow's PythonOperator + XCom."
Streaming story. Mage ships a first-class streaming pipeline mode with a Kafka / Kinesis / RabbitMQ source block, a transformer block (often powered by DuckDB, Materialize, or RisingWave for in-block SQL), and a sink block. Offsets are committed after the sink acknowledges — the exactly-once story most streaming orchestrators get wrong. Interviewers probe this because "batch and streaming in one tool" is a rare capability.
Integration coverage. Mage bundles 90+ pre-built connectors (Postgres, Snowflake, BigQuery, S3, GCS, Salesforce, Stripe, GitHub, HubSpot, MongoDB, etc.), a first-class dbt block (run dbt models as part of a Mage DAG), custom Python source / sink support, and — post-Airbyte-acquisition — a roadmap that pulls in the 350+ Airbyte connector catalogue. Interviewers probe this because it is the axis on which Mage wins for integration-heavy teams and loses to Dagster for lineage-heavy teams.

The 2026 reality — Mage is one of four production-grade orchestrators.

Mage AI. Open-source, Apache-2.0, notebook-native, block-based DAG, first-class streaming, 90+ integrations, dbt block, Airbyte-owned. Managed offering is Mage Pro. Wins for small-to-mid teams doing mixed batch + streaming + integration work who want dev-loop speed over lineage granularity.
Dagster. Open-source, Elementl-owned, software-defined asset (SDA) lineage-first, strong data quality integration, mature typing. Managed offering is Dagster Cloud. Wins for lineage-heavy analytics engineering where "which asset produced this row?" matters more than dev-loop speed.
Prefect. Open-source, Prefect-owned, dynamic Python flow model, first-class async, best-in-class retry and mapping. Managed offering is Prefect Cloud. Wins for dynamic, data-shape-dependent flows (fan-out over N files, per-tenant DAG shape) that a static DAG can't express.
Apache Airflow. ASF-hosted, largest ecosystem, mature scheduler, thousands of production deployments. Managed offerings from AWS MWAA, GCP Composer, Astronomer. Wins for legacy DAG scale and the "we've had this Airflow instance since 2018" reality; loses on dev-loop and streaming.

What interviewers listen for.

Do you name the block DAG model as the primary Mage differentiator, not "notebook UI"? — senior signal.
Do you distinguish the five block types (@data_loader, @transformer, @data_exporter, @sensor, @custom / scratchpad) without prompting? — required answer.
Do you name the Airbyte 2024 acquisition and the connector-catalogue roadmap convergence? — senior signal.
Do you describe the streaming pipeline as "Kafka source → windowed transform → sink → commit offsets" rather than as vague "streaming support"? — required answer.
Do you name Mage Pro as the managed tier and distinguish it from open-source Mage without prompting? — senior signal.

Worked example — the four-tool comparison table

Detailed explanation. The single most useful artifact for a Mage interview is a memorised 4x4 comparison table across Mage, Dagster, Prefect, and Airflow on the four axes interviewers care about. Every senior orchestrator discussion converges on this table within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical greenfield analytics platform.

Workload. Mixed: 40 batch ETL DAGs (Postgres → Snowflake), 3 streaming pipelines (Kafka clickstream → sessionise → Snowflake), 20 dbt models, 5 reverse-ETL flows (Snowflake → Salesforce).
Team. 6 data engineers, all Python-fluent, half also SQL-fluent, no dedicated platform engineer.
Requirements. Fast dev-loop (author → preview → schedule under 10 minutes), first-class dbt, streaming without a separate Flink cluster, secret management, on-prem or cloud.

Question. Build the four-tool comparison and pick the orchestrator each requirement drives you toward.

Input.

Tool	Dev-loop UX	Lineage (SDA)	Dynamic flows	Streaming	Community scale
Mage	live preview, browser IDE	block-level (implicit)	limited	first-class	mid
Dagster	UI materialisation	asset-graph (best)	good	via sensors	mid-large
Prefect	Python + subprocess	flow-run only	best (dynamic)	via workers	mid
Airflow	`airflow tasks test`	none native	limited (dynamic DAGs 2.3+)	via KafkaOperator	largest

Code.

# The Mage "hello block" — a one-file, three-block pipeline
# File: pipelines/hello_orders/load_orders.py
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd

@data_loader
def load_orders(*args, **kwargs) -> pd.DataFrame:
    """Return the latest 1000 orders — replace with your source."""
    return pd.read_sql(
        "SELECT id, customer_id, total_cents, status, updated_at "
        "FROM public.orders ORDER BY updated_at DESC LIMIT 1000",
        con=kwargs["context"]["postgres_conn"],
    )

# File: pipelines/hello_orders/transform_orders.py
from mage_ai.data_preparation.decorators import transformer
import pandas as pd

@transformer
def enrich(df: pd.DataFrame, *args, **kwargs) -> pd.DataFrame:
    df["total_usd"]   = df["total_cents"] / 100.0
    df["is_shipped"]  = df["status"].eq("shipped")
    return df

# File: pipelines/hello_orders/export_orders.py
from mage_ai.data_preparation.decorators import data_exporter
import pandas as pd

@data_exporter
def export(df: pd.DataFrame, *args, **kwargs) -> None:
    df.to_parquet(f"s3://acme-lake/orders/{kwargs['execution_date']}.parquet")

Step-by-step explanation.

The Mage pipeline is three files, one per block. Each file exports exactly one decorated function whose input signature (df: pd.DataFrame) implicitly declares its dependency on the upstream block's output. There is no separate DAG definition file — the DAG is the block-file wiring.
Compared to the Airflow equivalent — a dag.py that constructs a DAG(...) object, three PythonOperator tasks, and an xcom_pull boilerplate between them — the Mage version is ~30 lines vs Airflow's ~80, with no XCom coordination and no operator-vs-function ceremony.
Compared to Dagster's equivalent — three @asset decorators with explicit deps=[...] — Mage's implicit dependency (via the block file structure) is faster to author but less semantically rich for lineage tooling. Dagster's asset graph knows "orders_enriched depends on orders_raw"; Mage's DAG knows "block B runs after block A." That difference is why lineage-heavy teams still pick Dagster.
The dev-loop: open transform_orders.py in the Mage browser IDE, click Run, and the enriched DataFrame renders below the cell within seconds. This live-preview loop is what "notebook-style orchestrator" means in practice — the author never leaves the browser to test.
The pattern-match for the interview: "Mage is Airflow with the DAG-file-as-boilerplate replaced by decorator-per-block, plus a notebook IDE for interactive authoring." That sentence is the 10-word summary that scores highest.

Output.

Requirement	Winning tool	Why
Fast dev-loop (< 10 min author → schedule)	Mage	live preview; no DAG boilerplate
SDA lineage (which asset produced this row?)	Dagster	asset graph is a first-class object
Dynamic per-tenant flows	Prefect	dynamic mapping is the killer feature
Legacy DAG scale (5000+ DAGs)	Airflow	ecosystem + scheduler maturity
Mixed batch + streaming in one tool	Mage	streaming pipeline is first-class

Rule of thumb. Never pick an orchestrator based on "which one is trendy." Pick it based on (dev-loop × lineage × dynamic-flow × streaming × community). Write the 4x5 table on a whiteboard first; the tool falls out of the constraints.

Worked example — the "author → schedule" 10-minute dev-loop measurement

Detailed explanation. Every senior architect must quantify the dev-loop cost before signing off on an orchestrator, because that cost is paid by every engineer on every pipeline change for the next three years. The Mage dev-loop is measurably faster than Airflow's; measurably faster than Dagster's; and comparable to Prefect's for simple flows. Walk through a stopwatch-timed comparison for the same "load Postgres orders, enrich, write to S3" pipeline.

The task. Author a three-step pipeline, preview the middle transform's output, schedule it hourly, and see the first run succeed in the UI.
The clock. Wall-clock from "empty repo" to "green run in the UI."
The variables. Same engineer, same laptop, same source + sink, same day.

Question. Time the dev-loop for the same pipeline across Mage, Airflow, and Dagster and quantify the Mage advantage.

Input.

Step	Mage	Airflow	Dagster
Scaffold project	30 s (`mage start`)	90 s (init dags/, plugins/)	60 s (`dagster project scaffold`)
Write three blocks / tasks / assets	3 min	6 min (DAG boilerplate)	4 min (asset deps)
Preview middle step's output	15 s (live preview)	120 s (`airflow tasks test`)	60 s (materialise in UI)
Schedule (cron)	20 s (UI trigger)	60 s (schedule_interval + reload)	60 s (schedule object + reload)
First green run	30 s (immediate run)	90 s (scheduler pick-up)	60 s (run launcher)
Total	~4 min 35 s	~9 min 20 s	~6 min 20 s

Code.

# The Mage dev-loop — five commands from empty to running
pip install mage-ai
mage start acme_pipelines           # 1. spin up the browser IDE
# 2. open http://localhost:6789
# 3. click "New pipeline" → name it "hello_orders"
# 4. drag three blocks: data_loader, transformer, data_exporter
# 5. click Run — the pipeline runs; each block shows its output DataFrame

# Compare Airflow:
pip install apache-airflow
airflow db init
airflow users create --username admin --password admin --role Admin \
                     --email admin@acme.com --firstname A --lastname A
airflow scheduler &
airflow webserver &
# Now write dags/hello_orders.py by hand; test with:
airflow tasks test hello_orders load_orders 2026-07-21
airflow tasks test hello_orders transform_orders 2026-07-21
airflow tasks test hello_orders export_orders 2026-07-21
# Then trigger from the UI or `airflow dags trigger hello_orders`.

Step-by-step explanation.

Mage's scaffolding is one command (mage start) that both creates the project structure and launches the browser IDE. The engineer never leaves the browser thereafter — blocks are authored inline, run inline, scheduled inline.
Airflow's scaffolding requires db init, user creation, scheduler + webserver as separate processes, and the DAG file itself. The three tasks each need a PythonOperator(python_callable=...) binding with provide_context=True or the newer @task decorator, plus an XCom-based data hand-off between them.
Dagster's scaffolding is comparable to Airflow's in step count but with better ergonomics: dagster project scaffold, dagster dev, and three @asset decorators. The dev-loop is faster than Airflow (assets materialise from the UI) but slower than Mage (each materialisation is a separate UI action).
The measured difference (~4:35 for Mage vs ~9:20 for Airflow) is per pipeline change, not just per new pipeline. Over a year of iteration on 40 pipelines with (say) 5 edits per pipeline per month, the Mage delta is roughly 240 engineer-hours saved per year — enough to hire a junior engineer, or to build one more pipeline.
The trade-off is not free: Mage's speed comes from tight coupling between block author and block runtime. A 5000-DAG Airflow deployment survives multi-team development because the operator abstraction is stable and independently versioned; Mage's block-level decorators demand a shared Mage version across the team. This scales fine to ~10 engineers; less obviously to ~100.

Output.

Metric	Mage	Airflow	Dagster	Prefect
Empty-repo → first green run	~5 min	~9 min	~6 min	~5 min
Preview single step output	15 s	120 s	60 s	30 s
Author lines per 3-block pipeline	~30	~80	~50	~40
Interview signal	"fastest dev-loop"	"biggest ecosystem"	"best lineage"	"dynamic flows"

Rule of thumb. For any team where pipeline iteration velocity is the primary constraint, Mage is measurably faster than the alternatives. For any team where lineage granularity is the primary constraint, Dagster's asset graph is measurably richer. Pick the constraint that hurts most in production; the tool falls out of it.

Worked example — the "pick the orchestrator" decision tree

Detailed explanation. Given a new workload, the senior architect runs a 4-question decision tree in their head. Codifying the tree makes the interview answer reproducible: any interviewer can hand you a scenario and you can walk the tree out loud. Walk through the tree with three canonical scenarios: a greenfield analytics platform, a legacy Airflow-heavy team, and a lineage-obsessed analytics-engineering team.

Q1. Do you already have a large Airflow investment (>50 DAGs, dedicated platform team)? → yes = stay Airflow (migration cost > win); no = go to Q2.
Q2. Is SDA-level lineage a hard requirement (regulated data, "which asset touched this row?" audit)? → yes = Dagster; no = go to Q3.
Q3. Do your workloads need dynamic, data-shape-dependent flow structure (per-tenant DAG shape)? → yes = Prefect; no = go to Q4.
Q4. Do you need first-class streaming and dev-loop speed and rich integration coverage in one tool? → yes = Mage; no = re-evaluate — the workload might not need an orchestrator at all.
Q5 (parallel branch). Are you already deep in Airbyte for ingest? → yes = Mage becomes even stronger (post-2024 Airbyte acquisition roadmap convergence).

Question. Walk the decision tree for the three scenarios and record the tool each ends up with.

Input.

Scenario	Q1 (Airflow legacy?)	Q2 (SDA?)	Q3 (dynamic flows?)	Q4 (streaming + speed + integrations?)
Greenfield analytics platform	no	no	no	yes
Legacy Airflow (2018) team	yes	—	—	—
Regulated analytics-engineering	no	yes	—	—

Code.

# Decision-tree helper (illustrative)
def pick_orchestrator(has_airflow_legacy: bool,
                      needs_sda_lineage: bool,
                      needs_dynamic_flows: bool,
                      needs_streaming_speed_integrations: bool) -> str:
    """Return the recommended orchestrator for a new workload."""
    if has_airflow_legacy:
        return "Airflow (stay; migration cost > win)"
    if needs_sda_lineage:
        return "Dagster (asset graph is the killer feature)"
    if needs_dynamic_flows:
        return "Prefect (dynamic mapping is the killer feature)"
    if needs_streaming_speed_integrations:
        return "Mage (notebook UX + streaming + 90+ connectors)"
    return "Reconsider — the workload may not need an orchestrator"


# Walk the three scenarios
print(pick_orchestrator(False, False, False, True))
# → 'Mage (notebook UX + streaming + 90+ connectors)'

print(pick_orchestrator(True, False, False, False))
# → 'Airflow (stay; migration cost > win)'

print(pick_orchestrator(False, True, False, False))
# → 'Dagster (asset graph is the killer feature)'

Step-by-step explanation.

Scenario 1 — a greenfield analytics platform with mixed batch + streaming + 20 dbt models, 6 engineers, no dedicated platform team. Q1 = no, Q2 = no (lineage is nice-to-have), Q3 = no (flows are static), Q4 = yes → Mage. The notebook dev-loop plus first-class streaming plus dbt block plus 90+ connectors covers the entire workload.
Scenario 2 — a team with a 2018-vintage Airflow deployment running 300 DAGs, a dedicated platform team, and no burning pain-point besides "the DAG files are ugly." Q1 = yes → stay on Airflow. Migration cost dominates any dev-loop win; the correct answer is to fix the pain incrementally (TaskFlow API, dynamic DAGs 2.3+).
Scenario 3 — a regulated analytics-engineering team where "which asset touched this row?" is an audit requirement. Q1 = no, Q2 = yes → Dagster. The SDA graph is the load-bearing feature; Mage's block-level DAG is not semantically rich enough for audit-grade lineage.
The parallel Q5 (Airbyte) branch tips the balance further toward Mage: if your ingest is already Airbyte-managed, the post-2024 acquisition roadmap converges the Mage block model with the Airbyte connector catalogue, giving you one operational plane for connector runs and downstream transforms.
If none of Q1-Q4 pass, the correct answer is "you may not need an orchestrator at all" — a cron job on a small VM with a bash script can be the right answer for < 5 pipelines. Don't reflexively reach for orchestration.

Output.

Scenario	Recommended tool	Reason
Greenfield analytics platform	Mage	notebook + streaming + integrations
Legacy Airflow (2018) team	Airflow	migration cost > win
Regulated analytics-engineering	Dagster	SDA lineage requirement
Per-tenant dynamic flows	Prefect	dynamic mapping
< 5 pipelines total	cron on a VM	orchestration overhead not justified

Rule of thumb. The four-question decision tree is a whiteboard-friendly answer. Practice walking it end-to-end so an interviewer can hand you any scenario and get a tool name in under 60 seconds.

Senior interview question on Mage adoption

A senior interviewer often opens with: "Your team is on Airflow 2.6 with 40 DAGs and growing. Adoption has stalled — new engineers refuse to write DAG files because the boilerplate is painful. Should you migrate to Mage? Walk me through the evaluation criteria, the migration plan, the estimated cost, and the risks you'd guard against."

Solution Using a workload-by-workload migration plan anchored on dev-loop and streaming wins

# 1. Evaluation criteria — score each workload on 4 axes (0-3 each)
WORKLOADS = [
    # (name, dev_loop_win, streaming_win, integration_win, lineage_loss)
    ("nightly_postgres_to_snowflake",  3, 0, 2, 0),
    ("kafka_clickstream_sessionise",    2, 3, 1, 0),
    ("salesforce_to_snowflake_dbt",     3, 0, 3, 1),
    ("reverse_etl_snowflake_to_salesforce", 3, 0, 3, 1),
    ("regulated_finance_daily",         1, 0, 0, 3),  # keep on Airflow
]

def migration_score(w):
    _, dev, stream, integ, lineage = w
    return dev + stream + integ - lineage

# Rank
for w in sorted(WORKLOADS, key=migration_score, reverse=True):
    name, dev, stream, integ, lineage = w
    action = "MIGRATE" if migration_score(w) >= 4 else "KEEP"
    print(f"{action:8s} {name:38s}  score={migration_score(w)}")

# 2. Migration plan — three-phase rollout
phase_1_pilot:
  duration: 4 weeks
  workloads:
    - kafka_clickstream_sessionise   # unlocks Mage's streaming story
    - salesforce_to_snowflake_dbt    # unlocks Mage's dbt block
  success_criteria:
    - dev-loop wall-clock < 5 min author → run
    - streaming pipeline p99 latency < 30 s
    - zero missed dbt runs across pilot period

phase_2_bulk_migration:
  duration: 12 weeks
  workloads:
    - all batch ETL DAGs (Postgres → Snowflake × 20)
    - all reverse-ETL DAGs (Snowflake → Salesforce × 5)
  approach: convert PythonOperator tasks 1:1 to @data_loader / @transformer / @data_exporter blocks
  rollback: keep Airflow DAGs in `paused` state for 4 weeks post-migration

phase_3_decommission:
  duration: 4 weeks
  workloads: retire Airflow scheduler for the migrated set
  keep: `regulated_finance_daily` stays on Airflow (audit trail unchanged)

runtime_platform:
  option_a: self-host Mage on Kubernetes (helm chart)
  option_b: Mage Pro managed tier (SOC 2, priority support)

# 3. Airflow-to-Mage block adapter — port a PythonOperator in one file
# Before (Airflow):
# def load_orders(**context): return read_sql(...); return df.to_dict()
# t1 = PythonOperator(task_id='load_orders', python_callable=load_orders)

# After (Mage):
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd

@data_loader
def load_orders(*args, **kwargs) -> pd.DataFrame:
    return pd.read_sql(
        "SELECT id, customer_id, total_cents FROM public.orders WHERE updated_at > NOW() - INTERVAL '1 day'",
        con=kwargs["context"]["postgres_conn"],
    )

Step-by-step trace.

Step	Before (Airflow)	After (Mage)
Author time per new pipeline	~9 min	~5 min
Dev-loop per iteration	2 min (`airflow tasks test`)	15 s (live preview)
Streaming	separate Flink cluster	native Mage streaming block
dbt	BashOperator + dbt CLI	first-class dbt block
Integration coverage	operator-per-source	90+ pre-built connectors
Lineage	none native	block-level (implicit)
Team ramp	2-3 weeks	3-5 days

After deployment, the pilot two workloads land in production within four weeks; the streaming pipeline hits p99 latency of 18 s (target was 30 s); the dbt block runs 100% on schedule for the pilot period. Phase 2 converts the 40-DAG Airflow footprint into a 40-pipeline Mage footprint over 12 weeks; the regulated finance workload stays on Airflow because the audit tooling was built around Airflow's XCom + task-instance model.

Output:

Metric	Before (Airflow)	After (Mage)
Dev-loop per iteration	120 s	15 s
Author lines per 3-block pipeline	~80	~30
Streaming solution	separate Flink	in-tool
Team ramp for new engineer	2-3 weeks	3-5 days
Estimated engineer-hours saved / year	~240	(baseline)
Workloads migrated	0 / 40	39 / 40 (finance stays)

Why this works — concept by concept:

Workload-by-workload score — instead of "all Airflow → all Mage," score each workload on dev-loop win + streaming win + integration win minus lineage loss. Migrate only those with net-positive score. This preserves the audit trail on the regulated workloads while unlocking Mage on the rest.
Pilot on streaming + dbt — those are the two workloads where Mage's win is largest and Airflow's pain is worst. A successful pilot buys the political capital to do bulk migration in phase 2.
1:1 block port — porting PythonOperator to @data_loader is mechanical: the function body is unchanged, only the decorator and the return type. This keeps migration risk near zero.
Keep Airflow paused for 4 weeks post-migration — the safety net. If the Mage pipeline fails, un-pause the Airflow version and cut over back. Zero-downtime migration.
Cost — 4 weeks of pilot + 12 weeks of bulk + 4 weeks of decommission = 20 engineer-weeks total for a 40-DAG shop, offset by ~240 engineer-hours/year of dev-loop savings ongoing. Payback in ~6-9 months. The eliminated cost is the "new engineers refuse to write DAG files" adoption cliff. Net positive by year 1.

Design
Topic — design
Design problems on orchestrator selection and migration

Practice →

SQL Topic — sql SQL pipeline and ETL orchestration problems

Practice →

2. Block-based pipelines + DAG mental model

`@data_loader` → `@transformer` → `@data_exporter` — the block is the atom of every Mage DAG

The mental model in one line: mage blocks are the atomic unit of every Mage pipeline — each block is a Python (or SQL / R / PySpark) file with exactly one decorated function (@data_loader, @transformer, @data_exporter, @sensor, or @custom for scratchpad-style utilities), an input signature that implicitly declares its dependency on the outputs of its upstream blocks, and a return value that becomes the block's output — the DAG is not a separate file, it is the wiring graph of block files, and every block previews its output live in the browser IDE while you edit. Every senior data engineer who's built more than one Mage pipeline has internalised this model; the interview probe is whether you name the five block types unprompted and describe the implicit-dependency wiring.

The five block types — what each does and when to pick it.

@data_loader. Reads from a source; returns a DataFrame (or dict / list). Always the root of a pipeline sub-DAG — has no upstream. Typical sources: Postgres, S3, an HTTP API, a Kafka topic (in streaming mode). One pipeline can have multiple loaders that feed into a single transformer for join workloads.
@transformer. Reads from one or more upstream blocks (loaders or other transformers) via its input signature; returns a transformed DataFrame. This is the workhorse block — most business logic lives here. Supports Python (pandas / polars / DuckDB), SQL (against the current warehouse connection), R, and PySpark.
@data_exporter. Reads from an upstream transformer and writes to a sink — S3, Snowflake, Postgres, a Kafka topic, an HTTP endpoint. Typically the leaf of a pipeline sub-DAG; returns None (or, optionally, an assertion result).
@sensor. Blocks on an external condition — a file appears in S3, a Snowflake table exceeds a row count, a webhook fires. When the condition is met, downstream blocks are unblocked. Used for event-driven and inter-pipeline dependencies.
@custom (scratchpad). A block with no decorator restrictions — write arbitrary Python, run it once, throw it away. Used for one-off explorations that shouldn't be scheduled. The Mage version of the "let me just check something" cell.

The DAG wiring — implicit dependencies, no separate DAG file.

How the DAG is defined. In the browser IDE, drag-and-drop or explicitly declare that block B depends on block A. Mage writes this dependency into the pipeline's metadata.yaml (a single YAML file per pipeline that lists the blocks and their edges). You can edit metadata.yaml by hand — it's plain YAML — but the UI is the primary editor.
What's implicit. The upstream block's return value is passed as the first positional argument to the downstream block. If block A returns a DataFrame, block B's function signature is def b(df: pd.DataFrame, *args, **kwargs) — the df parameter is A's output. No XCom, no jinja, no explicit fetch.
Multiple upstreams. If block C has two upstreams A and B, its signature is def c(a_output, b_output, *args, **kwargs) — the positional order matches the dependency order in metadata.yaml.
DAG topology. Fan-out (one loader → three transformers) and fan-in (three loaders → one transformer for a three-way join) both work. Cycles are rejected at pipeline-save time.

Language support per block.

Python. Native, first-class, pandas / polars / DuckDB / PyArrow all work. This is the default.
SQL. A @transformer (or @data_loader) can be a .sql file with a Jinja-templated query. Mage runs it against the configured warehouse connection (Snowflake, BigQuery, Postgres, Redshift, etc.) and returns a DataFrame.
R. Same block-file structure with R decorators. Used by data-science teams that live in R.
PySpark. Blocks can run against a Spark session (attached externally or via a Mage-managed cluster). Used for big-data transforms that pandas can't hold.

Live preview — the dev-loop killer feature.

What it does. In the browser IDE, click Run on a block; Mage executes it against the current pipeline context and renders the return value inline — a DataFrame preview with the first N rows, a JSON pretty-print for dict returns, a plot for matplotlib returns.
Why it's fast. Mage caches upstream block outputs, so running a downstream block reuses the cache and executes only the changed block. Iterating on a transformer doesn't re-run the loader.
What it isn't. It is not a full Jupyter kernel — you cannot call arbitrary Python at the top level of a block file. The block is the function; imports and helpers go inside.

Git-versioning + block-file portability.

What's in Git. The pipelines/<name>/ directory: one file per block, plus metadata.yaml. All plain text. Diffs cleanly on pull requests.
What isn't in Git. Block output caches, run history, secrets. Those live in the Mage runtime state (a Postgres DB by default) and are managed separately.
Why this matters. A Mage pipeline is portable across environments — clone the repo, mage start, and every pipeline runs. No environment-specific DAG serialisation, no state migration.

Common interview probes on block DAG.

"Name the block types Mage supports." — required answer: @data_loader, @transformer, @data_exporter, @sensor, @custom (scratchpad).
"How does a downstream block receive its upstream's output?" — implicit positional argument on the decorated function.
"How is the DAG defined?" — the metadata.yaml per pipeline, plus block-file wiring; not a separate DAG.py.
"How does the live preview work?" — cached upstream outputs; runs only the changed block.

Worked example — three-block Postgres → transform → S3 pipeline

Detailed explanation. The canonical Mage batch pipeline: a data loader reads from Postgres, a transformer enriches the rows, a data exporter writes Parquet to S3. Walk through building it end-to-end — three block files, one metadata.yaml, one browser-UI Run.

Source. Postgres public.orders table.
Transform. Convert total_cents to total_usd, add an is_shipped boolean.
Sink. Parquet file on S3, keyed by execution date.

Question. Write the three block files and the metadata.yaml for the pipeline.

Input.

Component	Value
Source	Postgres `public.orders`
Transform	dollar conversion + is_shipped
Sink	S3 Parquet
Schedule	hourly
Runtime	Mage self-host or Mage Pro

Code.

# File: pipelines/orders_hourly/data_loaders/load_orders.py
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd
from sqlalchemy import create_engine

@data_loader
def load_orders(*args, **kwargs) -> pd.DataFrame:
    """Read the last hour of orders from Postgres."""
    engine = create_engine(kwargs["context"]["POSTGRES_URL"])
    return pd.read_sql(
        """
        SELECT id, customer_id, total_cents, status, updated_at
        FROM   public.orders
        WHERE  updated_at > NOW() - INTERVAL '1 hour'
        ORDER  BY updated_at
        """,
        con=engine,
    )

# File: pipelines/orders_hourly/transformers/enrich_orders.py
from mage_ai.data_preparation.decorators import transformer
import pandas as pd

@transformer
def enrich(df: pd.DataFrame, *args, **kwargs) -> pd.DataFrame:
    """Enrich orders with USD amount + is_shipped flag."""
    df = df.copy()
    df["total_usd"]  = df["total_cents"] / 100.0
    df["is_shipped"] = df["status"].eq("shipped")
    return df

# File: pipelines/orders_hourly/data_exporters/export_to_s3.py
from mage_ai.data_preparation.decorators import data_exporter
import pandas as pd

@data_exporter
def export(df: pd.DataFrame, *args, **kwargs) -> None:
    """Write to S3 as Parquet, keyed by execution date + hour."""
    ts       = kwargs["execution_date"].strftime("%Y/%m/%d/%H")
    s3_path  = f"s3://acme-lake/orders/{ts}.parquet"
    df.to_parquet(s3_path, index=False, compression="snappy")
    print(f"Wrote {len(df)} orders to {s3_path}")

# File: pipelines/orders_hourly/metadata.yaml
name: orders_hourly
uuid: orders_hourly
type: python
blocks:
  - uuid: load_orders
    name: load_orders
    type: data_loader
    language: python
    upstream_blocks: []
  - uuid: enrich_orders
    name: enrich_orders
    type: transformer
    language: python
    upstream_blocks:
      - load_orders
  - uuid: export_to_s3
    name: export_to_s3
    type: data_exporter
    language: python
    upstream_blocks:
      - enrich_orders

Step-by-step explanation.

Three block files in three subdirectories (data_loaders/, transformers/, data_exporters/) — Mage uses the subdirectory to type-check the decorator. Putting an @data_exporter-decorated function in data_loaders/ is a lint error.
load_orders has no upstream blocks (upstream_blocks: [] in the YAML) so its function signature is just *args, **kwargs. The kwargs["context"] dict carries per-run values including execution_date and the secrets you wired.
enrich_orders declares upstream_blocks: [load_orders], so its function signature is df: pd.DataFrame, *args, **kwargs — the df parameter is the return value of load_orders. Mage handles the pickling and passing; no XCom, no explicit fetch.
export_to_s3 declares upstream_blocks: [enrich_orders]. It returns None — data exporters typically write and return nothing. If they returned a value, downstream blocks (if any) would receive it.
In the browser IDE, opening any block file lets you click Run on that block. Running enrich_orders triggers load_orders (cache miss on first run, cache hit thereafter), passes its DataFrame to enrich_orders, and previews the enriched DataFrame inline. Iterating on the enrichment logic re-runs only the transformer.

Output.

Block	Input	Output	Runtime
load_orders	(none)	DataFrame with 1500 rows	~2 s
enrich_orders	1500 rows	same 1500 rows + 2 new cols	~50 ms
export_to_s3	1500 rows	S3 object (12 KB Parquet)	~800 ms

Rule of thumb. For every Mage pipeline, name your directories exactly data_loaders/, transformers/, data_exporters/, sensors/ — the subdirectory-based type check catches decorator mismatches at pipeline-save time. Declare upstream_blocks explicitly in metadata.yaml rather than relying on drag-and-drop; the YAML is the source of truth for the DAG shape.

Worked example — fan-in join of two loaders into one transformer

Detailed explanation. Most non-trivial pipelines have fan-in — two or more data loaders whose outputs are joined by a transformer. Mage handles this naturally: the downstream block's function signature takes one positional argument per upstream, in declaration order. Walk through a join of an orders loader and a customers loader into a joined-and-enriched transformer.

Loader A. Reads orders (id, customer_id, total_cents).
Loader B. Reads customers (id, name, tier).
Transformer. Joins on orders.customer_id = customers.id, emits enriched rows.

Question. Write the two-loader / one-transformer pipeline and demonstrate the multi-upstream function signature.

Input.

Component	Purpose
load_orders	Postgres `orders`
load_customers	Postgres `customers`
enrich_and_join	pandas `merge` on customer_id

Code.

# File: pipelines/orders_enriched/data_loaders/load_orders.py
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd
from sqlalchemy import create_engine

@data_loader
def load_orders(*args, **kwargs) -> pd.DataFrame:
    engine = create_engine(kwargs["context"]["POSTGRES_URL"])
    return pd.read_sql(
        "SELECT id, customer_id, total_cents FROM public.orders WHERE updated_at > NOW() - INTERVAL '1 day'",
        con=engine,
    )

# File: pipelines/orders_enriched/data_loaders/load_customers.py
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd
from sqlalchemy import create_engine

@data_loader
def load_customers(*args, **kwargs) -> pd.DataFrame:
    engine = create_engine(kwargs["context"]["POSTGRES_URL"])
    return pd.read_sql(
        "SELECT id AS customer_id, name, tier FROM public.customers",
        con=engine,
    )

# File: pipelines/orders_enriched/transformers/join_and_enrich.py
from mage_ai.data_preparation.decorators import transformer
import pandas as pd

@transformer
def join_and_enrich(orders: pd.DataFrame,
                    customers: pd.DataFrame,
                    *args, **kwargs) -> pd.DataFrame:
    """Fan-in join: orders + customers → enriched."""
    merged = orders.merge(customers, on="customer_id", how="left")
    merged["total_usd"] = merged["total_cents"] / 100.0
    return merged

# File: pipelines/orders_enriched/metadata.yaml
name: orders_enriched
blocks:
  - uuid: load_orders
    type: data_loader
    upstream_blocks: []
  - uuid: load_customers
    type: data_loader
    upstream_blocks: []
  - uuid: join_and_enrich
    type: transformer
    upstream_blocks:
      - load_orders           # first positional arg
      - load_customers        # second positional arg

Step-by-step explanation.

Two loaders run in parallel (Mage's scheduler schedules independent blocks concurrently). Each returns a DataFrame; each output is cached under its block UUID.
join_and_enrich declares two upstreams in upstream_blocks. The order in the YAML is the positional order of the function's parameters: orders is first (matches the first upstream load_orders), customers is second.
Getting the parameter order wrong (swapping load_orders and load_customers in the YAML but not in the function signature) is a common bug. Mage checks parameter count at runtime but not semantic meaning; the pandas merge would silently produce a bogus join. Always double-check upstream order after refactoring.
Fan-in is the right pattern for joins across independent sources. Fan-out (one loader → three transformers) is the right pattern for one source that feeds multiple sinks (e.g., write to Snowflake and to a Kafka topic and to a metrics endpoint). Both work naturally in the block model.
The live-preview loop is especially valuable for join development: run load_orders once, load_customers once, iterate on join_and_enrich many times — each iteration reuses the cached upstream outputs and completes in milliseconds.

Output.

Block	Rows in	Rows out	Runtime
load_orders (parallel)	(source)	5,200	1.2 s
load_customers (parallel)	(source)	2,800	0.9 s
join_and_enrich	5200 + 2800	5,200 (left join)	60 ms
Total wall-clock			1.3 s (loaders parallel)

Rule of thumb. For any fan-in Mage transformer, list the upstream blocks in metadata.yaml in the same positional order as the function signature's parameters. Double-check the mapping after every refactor — a swapped positional order silently produces the wrong join.

Worked example — SQL-language transformer against Snowflake

Detailed explanation. Mage transformers don't have to be Python. A .sql file with a Jinja-templated query, decorated implicitly by its file extension, runs against the configured warehouse connection and returns the query result as a DataFrame that downstream Python blocks can consume. Walk through a Snowflake-side aggregation transformer.

Source. A Python @data_loader that materialises a staging Snowflake table.
Transform. A SQL @transformer that aggregates via GROUP BY in Snowflake.
Sink. A Python @data_exporter that writes to S3.

Question. Add a SQL-language transformer to the pipeline and show how it exchanges data with Python-language blocks upstream and downstream.

Input.

Component	Language	Purpose
stage_orders	Python	write to Snowflake staging
aggregate_by_tier	SQL	GROUP BY in Snowflake
export_summary	Python	S3 Parquet write

Code.

# File: pipelines/orders_summary/data_loaders/stage_orders.py
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd
from snowflake.connector.pandas_tools import write_pandas
import snowflake.connector

@data_loader
def stage_orders(*args, **kwargs) -> pd.DataFrame:
    """Write orders to a Snowflake staging table; return the row count."""
    df = pd.read_csv("s3://acme-lake/orders/latest.csv")
    conn = snowflake.connector.connect(**kwargs["context"]["SNOWFLAKE_CONN"])
    write_pandas(conn, df, "orders_stage", schema="STAGING", overwrite=True)
    return df    # downstream can also use the DataFrame directly if it wants

-- File: pipelines/orders_summary/transformers/aggregate_by_tier.sql
-- Mage auto-runs this against the configured Snowflake profile
SELECT
    c.tier                            AS customer_tier,
    COUNT(*)                          AS n_orders,
    SUM(o.total_cents) / 100.0        AS revenue_usd,
    AVG(o.total_cents) / 100.0        AS avg_order_usd
FROM   STAGING.orders_stage o
JOIN   analytics.dim_customers c
  ON   c.id = o.customer_id
GROUP  BY c.tier
ORDER  BY revenue_usd DESC;

# File: pipelines/orders_summary/data_exporters/export_summary.py
from mage_ai.data_preparation.decorators import data_exporter
import pandas as pd

@data_exporter
def export_summary(summary: pd.DataFrame, *args, **kwargs) -> None:
    """Write the aggregated summary to S3."""
    ts   = kwargs["execution_date"].strftime("%Y%m%d")
    path = f"s3://acme-lake/summaries/orders_by_tier_{ts}.parquet"
    summary.to_parquet(path, index=False)

# File: pipelines/orders_summary/metadata.yaml
name: orders_summary
data_integration:
  destination_profile: snowflake_prod
blocks:
  - uuid: stage_orders
    type: data_loader
    language: python
    upstream_blocks: []
  - uuid: aggregate_by_tier
    type: transformer
    language: sql
    configuration:
      data_provider: snowflake
      data_provider_profile: default
    upstream_blocks:
      - stage_orders
  - uuid: export_summary
    type: data_exporter
    language: python
    upstream_blocks:
      - aggregate_by_tier

Step-by-step explanation.

The .sql file's extension declares the block's language. Mage's SQL runtime executes the query against the profile named in configuration.data_provider_profile (a connection string configured in the Mage runtime settings, not committed to Git).
Data flows across language boundaries via DataFrames. The Python stage_orders returns a DataFrame and writes to Snowflake staging. The SQL transformer reads from that Snowflake table and returns its result set as a DataFrame. The Python export_summary consumes that DataFrame.
The SQL transformer's SELECT result set becomes the block's output DataFrame — no INSERT INTO, no side-effect writes. This is important: SQL blocks are transforms, not sinks. If you want to write to Snowflake, use a data_exporter block with SQL or a Python data_exporter using the Snowflake connector.
Jinja templating is supported in SQL blocks: {{ execution_date }}, {{ kwargs['some_var'] }}, and custom filters all work. This lets you write parametric SQL (e.g., WHERE order_date = '{{ execution_date }}') without a Python wrapper.
Mixing languages per block is the killer feature for analytics-engineering teams: the "shape the data in SQL, ship it in Python" workflow that used to require two tools now fits in one pipeline. dbt users will recognise this as a lightweight dbt — for full dbt integration, use the dedicated dbt block type (covered in section 4).

Output.

Block	Language	Runtime	Rows out
stage_orders	Python	4 s	12,000
aggregate_by_tier	SQL (Snowflake)	1.8 s	4 (one per tier)
export_summary	Python	0.4 s	(writes 4-row Parquet)

Rule of thumb. For any transformation that would compile to a GROUP BY or a WINDOW function, prefer a SQL block over a pandas block — the warehouse is faster, cheaper (no data egress), and the query optimiser is smarter. Use pandas for row-level logic that SQL can't express or that needs a Python-only library.

Senior interview question on block DAG design

A senior interviewer might ask: "Design a Mage pipeline that ingests three source systems (Postgres, S3, Salesforce), joins them into a canonical customer_360 table, writes to Snowflake, and triggers a downstream dbt model. Explain the block topology, the upstream declarations, the SQL / Python language choices per block, and the failure semantics when one of the source loaders fails."

Solution Using a fan-in loader graph, a SQL join transformer, and a dbt block for the downstream model

# 1. Pipeline metadata — the DAG shape
name: customer_360
uuid: customer_360
type: python
blocks:
  # Three parallel loaders — fan-out at the source
  - uuid: load_postgres_customers
    type: data_loader
    language: python
    upstream_blocks: []
  - uuid: load_s3_orders
    type: data_loader
    language: python
    upstream_blocks: []
  - uuid: load_salesforce_contacts
    type: data_loader
    language: python
    upstream_blocks: []

  # Stage each source into Snowflake so the join runs warehouse-side
  - uuid: stage_customers_to_snowflake
    type: transformer
    language: python
    upstream_blocks:
      - load_postgres_customers
  - uuid: stage_orders_to_snowflake
    type: transformer
    language: python
    upstream_blocks:
      - load_s3_orders
  - uuid: stage_contacts_to_snowflake
    type: transformer
    language: python
    upstream_blocks:
      - load_salesforce_contacts

  # SQL transformer joins in the warehouse
  - uuid: join_customer_360
    type: transformer
    language: sql
    configuration:
      data_provider: snowflake
      data_provider_profile: default
    upstream_blocks:
      - stage_customers_to_snowflake
      - stage_orders_to_snowflake
      - stage_contacts_to_snowflake

  # dbt block triggers downstream models
  - uuid: run_dbt_customer_360
    type: dbt
    configuration:
      dbt_project_path: /home/mage/dbt/acme
      command: run
      select: tag:customer_360
    upstream_blocks:
      - join_customer_360

# 2. Loader — Salesforce (illustrative, uses Mage's Salesforce connector)
from mage_ai.data_preparation.decorators import data_loader
from mage_ai.io.salesforce import Salesforce
import pandas as pd

@data_loader
def load_salesforce_contacts(*args, **kwargs) -> pd.DataFrame:
    """Pull contacts modified in the last 24 hours."""
    client = Salesforce.with_config(kwargs["context"]["SF_CONFIG"])
    soql   = ("SELECT Id, Email, AccountId, LastModifiedDate "
              "FROM Contact WHERE LastModifiedDate > LAST_N_DAYS:1")
    return client.query(soql)

-- 3. SQL transformer — join in Snowflake, avoid egress
CREATE OR REPLACE TEMPORARY TABLE customer_360_stage AS
SELECT
    c.id                     AS customer_id,
    c.name,
    c.tier,
    sc.email                 AS sf_email,
    sc.account_id            AS sf_account_id,
    COUNT(o.id)              AS lifetime_orders,
    SUM(o.total_cents) / 100 AS lifetime_revenue_usd,
    MAX(o.updated_at)        AS last_order_at
FROM   STAGING.customers    c
LEFT JOIN STAGING.orders    o  ON o.customer_id = c.id
LEFT JOIN STAGING.contacts  sc ON sc.email      = c.email
GROUP  BY c.id, c.name, c.tier, sc.email, sc.account_id;

SELECT * FROM customer_360_stage;

Step-by-step trace.

Layer	Block	Runs on	Role
Fan-out load	3 parallel loaders	Mage runtime (Python)	source ingestion
Stage	3 stagers	Mage runtime → Snowflake	move data to warehouse
Fan-in join	SQL block	Snowflake warehouse	GROUP BY in warehouse
Downstream	dbt block	Mage runtime → dbt CLI	run dbt models tagged customer_360
Failure semantics	any loader fails	Mage retries per block	join blocked; dbt not triggered

After deployment, the three loaders run in parallel in ~3 s wall clock, staging takes ~5 s total, the Snowflake join runs in ~1.2 s, and dbt models take ~2 min. End-to-end p99 ~2 min 12 s. A loader failure (e.g., Salesforce API rate limit) fails that block only; downstream join_customer_360 waits for the retry; if all retries exhaust, the join and dbt block are skipped and the run is marked failed.

Output:

Metric	Value
Wall-clock (steady state)	~2 min 12 s
Parallel loader speedup	3x (vs sequential)
Warehouse-side join savings	~15 s (no egress)
dbt integration	native block, not BashOperator
Failure surface (per block)	Mage retries with exponential backoff
Rerun granularity	per block (skip completed)

Why this works — concept by concept:

Fan-out loaders — three loaders with upstream_blocks: [] run in parallel because Mage's scheduler schedules independent blocks concurrently. This is a 3x wall-clock win over a sequential loader chain.
Stage-then-join in the warehouse — the SQL join runs inside Snowflake, so the terabytes of orders never leave the warehouse. This is the same "push down the join" logic dbt teams internalise; Mage lets you express it as a single-block transform.
dbt block — type: dbt is a first-class Mage block that shells out to dbt run --select tag:customer_360 under the hood, but with Mage's scheduler wrapping it (retries, logging, notifications). This is qualitatively better than an Airflow BashOperator because Mage understands the dbt model list and can report per-model status.
Failure semantics per block — Mage retries each block independently. A Salesforce API rate-limit failure retries the loader with exponential backoff; the join block waits. If the loader fails permanently, the run is marked failed and no downstream block runs. This is the standard orchestrator failure model but with block-level granularity.
Cost — 8 blocks total, ~2 min p99 wall clock, native retries, native dbt integration, warehouse-side join for cost efficiency. Compared to the Airflow equivalent (PythonOperator × 3 loaders + PythonOperator × 3 stagers + SnowflakeOperator + BashOperator for dbt + inter-task XCom) this is ~40% less code and dramatically better failure ergonomics.

SQL
Topic — sql
SQL fan-in join and aggregation problems

Practice →

Aggregation Topic — aggregation Aggregation and GROUP BY problems

Practice →

3. Streaming pipelines in Mage

Kafka source block → windowed transform → sink → commit offsets — the exactly-once streaming story most orchestrators get wrong

The mental model in one line: mage streaming is Mage's first-class streaming pipeline mode where a source block (Kafka, Kinesis, RabbitMQ, Amazon SQS, Google Pub/Sub) reads a message log, a transformer block (often powered by DuckDB, Materialize, or RisingWave for in-block SQL windowing) shapes each message batch, a sink block (Kafka, Kinesis, Snowflake, BigQuery, S3, Postgres) commits the transformed output, and — critically — the source offsets are committed after the sink acknowledges, giving you at-least-once semantics by default and effectively exactly-once when the sink is idempotent, all inside the same block DAG runtime that runs your batch pipelines. This is the axis where Mage differentiates most sharply from Airflow (which needs KafkaOperator boilerplate) and Prefect (which requires an out-of-band worker).

Streaming pipeline mode — how it differs from batch.

Pipeline type. In metadata.yaml, type: streaming (vs type: python). This flag changes the executor: the pipeline runs continuously (not per schedule), the source block polls the message queue in a loop, and each polled batch flows through the transformer(s) and sink(s) before the source offset advances.
Block types. Same five block types, but data_loader is renamed conceptually to source (still uses @streaming_source decorator), and data_exporter to sink (@streaming_sink). Transformers are unchanged.
Execution model. Continuous loop: pull batch → transform → sink → commit offset → repeat. The block DAG is the unit of processing per batch, not per run.
Batching. The source block yields batches (100-1000 messages per batch is typical). The whole pipeline processes one batch atomically before the next.

Kafka source block — the canonical streaming source.

Config. bootstrap.servers, topic, group.id, auto.offset.reset (earliest for backfill, latest for tailing), enable.auto.commit=false (Mage commits explicitly after sink).
Batch size. max.poll.records = 500 is a common default. Larger batches amortise fixed overhead; smaller batches reduce end-to-end latency.
Deserialisation. JSON, Avro (with Confluent Schema Registry), Protobuf. Mage wraps the standard kafka-python / confluent-kafka clients.
Exactly-once posture. Mage commits Kafka offsets after the sink acknowledges. If the sink is idempotent (Snowflake MERGE, S3 with deterministic keys), the pipeline is effectively exactly-once. If not, it is at-least-once — duplicate delivery on failure is possible.

In-block window / aggregation engines — DuckDB, Materialize, RisingWave.

DuckDB. An embedded analytical engine. Mage transformers can import duckdb and run SQL over the current batch's DataFrame — perfect for per-batch aggregations, sessionisation, or windowed joins that pandas would struggle with.
Materialize. A streaming SQL engine. Mage can point a transformer at a Materialize source and let Materialize do the incremental view maintenance; the block reads the materialised view.
RisingWave. A newer streaming SQL engine, similar posture to Materialize but with a distinct incremental-materialisation model. Mage supports both as engine-in-the-block partners.
When to pick which. DuckDB for stateless per-batch transforms (fastest, no extra service). Materialize/RisingWave for stateful cross-batch windows that must survive restarts (the streaming engine owns the state).

Window and aggregation blocks — the state story.

Stateless per-batch. Group by key inside the batch; emit one row per key. No cross-batch state; trivial to reason about.
Tumbling window. Fixed-size time windows (e.g., 1-minute tumbling). State lives inside the transformer for the duration of one window; emitted on window close.
Sliding / hopping window. Overlapping windows (e.g., 5-min window advancing every 1 min). Needs more state; usually delegated to Materialize / RisingWave rather than in-Python.
Session window. Windows defined by inactivity gaps (e.g., "session = 30 min of no events"). The most stateful pattern; strongly prefer a dedicated engine.

Checkpointing and offset semantics.

The commit dance. Mage's default flow: (1) poll batch from Kafka, (2) run transformer(s), (3) call sink, (4) only after sink returns success, commit the Kafka consumer offset. Failure at step 2 or 3 → offset not committed → the same batch is re-polled on retry.
The idempotency requirement on sinks. Because retries re-deliver the same batch, sinks must be idempotent. Snowflake MERGE INTO ... USING VALUES ... is idempotent. S3 with deterministic keys (s3://bucket/YYYY/MM/DD/HH/batch_id.parquet) is idempotent. Kafka with keyed messages + log compaction is idempotent. Postgres INSERT ... ON CONFLICT DO UPDATE is idempotent. Pick one.
Backfill vs tail. auto.offset.reset=earliest on first start = backfill the entire topic. latest = tail only new messages. Once the consumer group has committed offsets, this flag is ignored (Kafka uses the committed position).

Common interview probes on Mage streaming.

"How does Mage's streaming pipeline differ from a batch pipeline?" — required answer: continuous loop, type: streaming, source polls messages, offsets committed after sink.
"What's the exactly-once story?" — required answer: at-least-once by default; effectively exactly-once when sinks are idempotent.
"When would you pick Materialize inside a Mage transformer vs a dedicated Flink cluster?" — Mage + Materialize for small-to-mid streaming (< 100k events/sec); Flink for very large state or complex event processing.
"How do you handle back-pressure?" — Mage's poll-transform-sink loop naturally back-pressures (the source doesn't poll faster than the sink can absorb); tune max.poll.records to bound batch size.

Worked example — Kafka clickstream sessionisation → Snowflake

Detailed explanation. The canonical Mage streaming pipeline: a Kafka source reads a clickstream topic, an in-block DuckDB SQL transformer sessionises the events (30-min inactivity gap), a Snowflake sink writes one row per completed session. Walk through the three blocks and the metadata.yaml.

Source. Kafka topic clickstream.raw, JSON messages with {user_id, ts, page}.
Transform. DuckDB SQL to sessionise: for each user, group events into sessions where inactivity > 30 min splits the session.
Sink. Snowflake analytics.sessions — one row per session with user_id, session_id, start_ts, end_ts, page_count.

Question. Write the source, transformer, and sink blocks plus the streaming metadata.yaml.

Input.

Component	Value
Kafka topic	clickstream.raw
Batch size	500 messages
Session window	30-min inactivity
Snowflake table	analytics.sessions (MERGE by session_id)
Offset commit	after sink ack

Code.

# File: pipelines/clickstream_sessions/data_loaders/kafka_source.py
from mage_ai.streaming.sources.kafka import KafkaSource
from mage_ai.data_preparation.decorators import streaming_source

@streaming_source
def source(*args, **kwargs) -> KafkaSource:
    """Kafka source for the clickstream.raw topic."""
    return KafkaSource(config=dict(
        bootstrap_servers   = kwargs["context"]["KAFKA_BROKERS"],
        topic               = "clickstream.raw",
        consumer_group      = "mage-clickstream-sessioniser",
        auto_offset_reset   = "latest",
        enable_auto_commit  = False,
        max_poll_records    = 500,
        serde               = "json",
    ))

# File: pipelines/clickstream_sessions/transformers/sessionise.py
from mage_ai.data_preparation.decorators import transformer
import duckdb
import pandas as pd

@transformer
def sessionise(batch: list[dict], *args, **kwargs) -> pd.DataFrame:
    """Sessionise the batch: 30-min inactivity gap defines a new session."""
    df = pd.DataFrame(batch)
    if df.empty:
        return df

    con = duckdb.connect(":memory:")
    con.register("events", df)
    result = con.execute("""
        WITH ordered AS (
            SELECT
                user_id, ts, page,
                LAG(ts) OVER (PARTITION BY user_id ORDER BY ts) AS prev_ts
            FROM events
        ),
        gaps AS (
            SELECT *,
                CASE WHEN prev_ts IS NULL
                       OR ts - prev_ts > INTERVAL '30 minutes'
                     THEN 1 ELSE 0 END AS is_new_session
            FROM ordered
        ),
        sessioned AS (
            SELECT *,
                SUM(is_new_session) OVER (PARTITION BY user_id ORDER BY ts) AS session_idx
            FROM gaps
        )
        SELECT
            user_id,
            user_id || '_' || CAST(session_idx AS VARCHAR) AS session_id,
            MIN(ts)  AS start_ts,
            MAX(ts)  AS end_ts,
            COUNT(*) AS page_count
        FROM sessioned
        GROUP BY user_id, session_idx
    """).df()
    return result

# File: pipelines/clickstream_sessions/data_exporters/snowflake_sink.py
from mage_ai.data_preparation.decorators import streaming_sink
import snowflake.connector
import pandas as pd

@streaming_sink
def sink(sessions: pd.DataFrame, *args, **kwargs) -> None:
    """MERGE the completed sessions into analytics.sessions — idempotent."""
    if sessions.empty:
        return
    conn = snowflake.connector.connect(**kwargs["context"]["SNOWFLAKE_CONN"])
    cur  = conn.cursor()
    # Stage the batch
    cur.execute("CREATE TEMPORARY TABLE stage_sessions LIKE analytics.sessions")
    from snowflake.connector.pandas_tools import write_pandas
    write_pandas(conn, sessions, "STAGE_SESSIONS", schema="analytics", overwrite=True)
    # Idempotent MERGE
    cur.execute("""
        MERGE INTO analytics.sessions tgt
        USING analytics.stage_sessions src
          ON tgt.session_id = src.session_id
        WHEN MATCHED THEN UPDATE SET
            end_ts     = src.end_ts,
            page_count = src.page_count
        WHEN NOT MATCHED THEN INSERT (user_id, session_id, start_ts, end_ts, page_count)
                          VALUES (src.user_id, src.session_id, src.start_ts, src.end_ts, src.page_count)
    """)
    conn.commit()
    conn.close()

# File: pipelines/clickstream_sessions/metadata.yaml
name: clickstream_sessions
uuid: clickstream_sessions
type: streaming
blocks:
  - uuid: kafka_source
    type: data_loader
    language: python
    upstream_blocks: []
  - uuid: sessionise
    type: transformer
    language: python
    upstream_blocks:
      - kafka_source
  - uuid: snowflake_sink
    type: data_exporter
    language: python
    upstream_blocks:
      - sessionise

Step-by-step explanation.

type: streaming in metadata.yaml switches Mage to the continuous execution model. The pipeline starts once and runs forever (or until stopped); no cron schedule.
The @streaming_source returns a KafkaSource object, not a DataFrame. Mage's streaming runtime calls .poll() on it in a loop, each poll returning up to 500 messages as a list[dict]. Setting enable_auto_commit=False is mandatory — Mage must own the commit to guarantee post-sink semantics.
The transformer receives the batch as a list[dict] (raw Kafka message payloads), converts to a DataFrame, and runs DuckDB SQL for the sessionisation. DuckDB inside a Python block is the "cheat code" of Mage streaming: full SQL windowing power without a separate streaming engine.
The sink runs a Snowflake MERGE — critically, MERGE is idempotent, so re-delivering the same batch on failure produces the same end-state. Combined with post-sink offset commit, this is effectively exactly-once.
On failure at any point (transformer exception, sink error), Mage does not commit the Kafka offset. The same batch is re-polled on the next iteration. Combined with the idempotent sink, this yields zero data loss and no duplicates.

Output.

Stage	Input	Output	Latency
Kafka poll	(offset 12,340,000)	500 msgs	~50 ms
Sessionise (DuckDB)	500 msgs	47 sessions	~15 ms
Snowflake MERGE	47 sessions	(upsert)	~180 ms
Offset commit	(offset 12,340,500)	(commit ack)	~10 ms
End-to-end per batch			~255 ms

Rule of thumb. For every Mage streaming pipeline, always set enable_auto_commit=False on Kafka sources, always use an idempotent sink pattern (Snowflake MERGE, S3 deterministic keys, Postgres ON CONFLICT), and never manually commit offsets inside the sink — Mage's streaming runtime does that after the sink returns success.

Worked example — DuckDB tumbling-window aggregation

Detailed explanation. A common streaming shape: aggregate events into fixed time windows (say, 1-minute tumbling) and emit one row per (window, key). Mage transformers can hold cross-batch state in a module-level dict; DuckDB inside the transformer handles the window logic. Walk through a per-user page-view count.

Source. Kafka topic pageviews.raw (500 msgs/batch).
Transform. 1-minute tumbling window; per-user page count.
Sink. BigQuery analytics.pageviews_1min.

Question. Write a transformer that buffers events across batches and emits one row per window on window close.

Input.

Component	Value
Window	1-min tumbling
Key	user_id
Aggregation	COUNT(*)
Buffer	in-memory dict keyed by window start
Emit trigger	window closes when ts > window_end

Code.

# File: pipelines/pageviews_1min/transformers/tumbling_1min.py
from mage_ai.data_preparation.decorators import transformer
import duckdb
import pandas as pd
from collections import defaultdict
from datetime import datetime, timedelta

# Module-level state — Mage keeps the transformer object alive across batches
_buffer: dict[datetime, list[dict]] = defaultdict(list)

def _window_start(ts: datetime) -> datetime:
    """Round timestamp DOWN to the nearest 1-minute boundary."""
    return ts.replace(second=0, microsecond=0)

@transformer
def tumbling_1min(batch: list[dict], *args, **kwargs) -> pd.DataFrame:
    """Bucket events into 1-min windows; emit only closed windows."""
    now = datetime.utcnow()
    for msg in batch:
        ts = datetime.fromisoformat(msg["ts"])
        _buffer[_window_start(ts)].append(msg)

    # A window is "closed" when its end (start + 1 min) is < now - 30s watermark
    watermark = now - timedelta(seconds=30)
    closed_windows = [w for w in list(_buffer.keys()) if w + timedelta(minutes=1) < watermark]
    if not closed_windows:
        return pd.DataFrame()

    rows = []
    for w in closed_windows:
        df = pd.DataFrame(_buffer.pop(w))
        agg = (df.groupby("user_id")
                 .size()
                 .reset_index(name="page_count"))
        agg["window_start"] = w
        agg["window_end"]   = w + timedelta(minutes=1)
        rows.append(agg)

    return pd.concat(rows, ignore_index=True)

# File: pipelines/pageviews_1min/data_exporters/bigquery_sink.py
from mage_ai.data_preparation.decorators import streaming_sink
from google.cloud import bigquery
import pandas as pd

@streaming_sink
def sink(df: pd.DataFrame, *args, **kwargs) -> None:
    """Append closed windows to BigQuery — idempotent by (window_start, user_id)."""
    if df.empty:
        return
    client = bigquery.Client(project=kwargs["context"]["GCP_PROJECT"])
    table  = "analytics.pageviews_1min"
    # MERGE via load job with a WRITE_APPEND + downstream dedup, or explicit MERGE
    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_APPEND",
        schema=[
            bigquery.SchemaField("user_id", "STRING"),
            bigquery.SchemaField("page_count", "INT64"),
            bigquery.SchemaField("window_start", "TIMESTAMP"),
            bigquery.SchemaField("window_end",   "TIMESTAMP"),
        ],
    )
    client.load_table_from_dataframe(df, table, job_config=job_config).result()

Step-by-step explanation.

_buffer is a module-level dict — because Mage keeps the transformer's Python module loaded across batches, module-level state persists. This is the simplest way to hold cross-batch state without an external store.
Every incoming message is bucketed into its 1-minute window (rounded down). Buckets accumulate as batches arrive; a bucket is not emitted until its window has "closed" (window_end < now - 30 s watermark, giving 30 s of slack for late-arriving events).
When a window closes, the transformer pops it from _buffer, runs a per-user COUNT via pandas, adds window_start and window_end columns, and returns the aggregated rows. Older batches whose buckets are still open contribute additional rows to those buckets but don't cause emission.
The watermark (30 s) is a trade-off: longer = more forgiving of late events, but higher end-to-end latency; shorter = tighter latency, but risks late events being dropped. 30 s is a good default for well-behaved clickstreams; go up to a few minutes for messy sources.
The BigQuery sink uses WRITE_APPEND — because Mage may re-emit the same window on failure, the downstream table needs periodic deduplication (or use a MERGE with session_id as the key). The (window_start, user_id) composite is naturally unique per window, so dedup is cheap.

Output.

Batch #	Buffer state	Closed windows	Rows emitted
1	3 open windows	0	0
2	3 open windows	0	0
3	2 open windows	1 closed	47
4	2 open windows	0	0
5	3 open windows	1 closed	52

Rule of thumb. For any Mage streaming tumbling / sliding window, hold buffer state in a module-level dict, define a watermark (~30 s), and emit only closed windows. For sessions longer than an hour, or for very high cardinality keys, use Materialize or RisingWave via a dedicated block — the in-Python approach doesn't scale to millions of open sessions.

Worked example — when to pick Mage streaming vs a dedicated Flink cluster

Detailed explanation. The senior architect trade-off: Mage's in-runtime streaming is dramatically simpler than standing up Flink or Spark Streaming, but has a scaling ceiling. Walk through the four axes that decide the pick.

Throughput. Mage streaming handles ~50-100k events/sec on a single worker. Flink handles millions/sec across a cluster.
State size. Mage's in-Python state is bounded by worker memory (~ few GB). Flink's RocksDB state backend handles TB-scale state with local disk + checkpoints to S3.
Complex event processing. Flink CEP is best-in-class; Mage has no native CEP.
Operational simplicity. Mage runs in the same runtime as your batch pipelines. Flink is a separate cluster, JobManager, TaskManagers, state backend, savepoint operations.

Question. Given four hypothetical workloads, pick Mage streaming or Flink for each and justify.

Input.

Workload	Throughput	State size	Complex CEP?	Team Flink expertise?
Clickstream sessionisation	20k/s	~500 MB (open sessions)	no	no
IoT event enrichment	200k/s	~10 GB (device state)	no	yes
Fraud detection multi-pattern	5k/s	~2 GB	yes (CEP)	yes
Simple Kafka → Snowflake copy	10k/s	none	no	no

Code.

# Decision helper
def pick_streaming_platform(throughput_per_sec: int,
                             state_gb: float,
                             needs_cep: bool,
                             team_knows_flink: bool) -> str:
    if throughput_per_sec > 100_000:
        return "Flink (throughput above Mage single-worker ceiling)"
    if state_gb > 5:
        return "Flink (RocksDB state backend needed for TB-scale)"
    if needs_cep:
        return "Flink CEP (Mage has no native CEP library)"
    if not team_knows_flink and throughput_per_sec < 50_000 and state_gb < 2:
        return "Mage streaming (simpler; team knows Mage)"
    return "Mage streaming (safe default under thresholds)"

print(pick_streaming_platform(20_000,   0.5, False, False))
# → 'Mage streaming (simpler; team knows Mage)'
print(pick_streaming_platform(200_000, 10.0, False, True))
# → 'Flink (throughput above Mage single-worker ceiling)'
print(pick_streaming_platform(5_000,    2.0, True,  True))
# → 'Flink CEP (Mage has no native CEP library)'
print(pick_streaming_platform(10_000,   0.0, False, False))
# → 'Mage streaming (simpler; team knows Mage)'

Step-by-step explanation.

Workload 1 (clickstream, 20k/s, 500 MB state, no CEP, no Flink) → Mage. Below the throughput ceiling, below the state ceiling, no CEP needed, team doesn't want to operate Flink. This is the sweet spot.
Workload 2 (IoT, 200k/s) → Flink. Above Mage's single-worker throughput ceiling. Even though the team knows Flink, the driving factor is raw throughput.
Workload 3 (fraud, 5k/s but CEP) → Flink. Low throughput fits Mage, but CEP semantics (pattern matching across event sequences) are Flink's strength. Mage would require hand-rolling CEP, which is a maintenance nightmare.
Workload 4 (simple copy, 10k/s) → Mage. Trivially fits Mage streaming; running Flink for a straight copy is over-engineering.
The pattern: default to Mage under ~50k/s + < 2 GB state + no CEP. Above any of those thresholds, evaluate Flink honestly. The operational cost of Flink is significant; don't pay it without a matching workload.

Output.

Workload	Pick	Reason
Clickstream sessionisation	Mage	fits every threshold
IoT enrichment	Flink	throughput ceiling
Fraud detection CEP	Flink	CEP semantics
Simple Kafka → Snowflake	Mage	over-engineering to run Flink

Rule of thumb. Default to Mage streaming for any workload under ~50k events/sec with < 2 GB of state and no CEP requirement. Escalate to Flink when any of those thresholds is exceeded. The Mage-streaming operational win vanishes above the ceiling; the Flink operational cost vanishes when the workload genuinely needs it.

Senior interview question on Mage streaming

A senior interviewer might ask: "You have a Kafka topic orders.events producing 30k messages/sec. Downstream needs a per-user 5-minute tumbling window count into Snowflake for real-time dashboards, exactly-once semantics. Design the Mage streaming pipeline, including the source config, the DuckDB window transformer, the Snowflake idempotent sink, the offset-commit ordering, and the failure semantics under a 10-minute Kafka broker outage."

Solution Using a Kafka source + module-state windowed transformer + Snowflake MERGE sink + post-sink offset commit

# 1. Kafka source — 500 msg/batch, no auto-commit, JSON
from mage_ai.streaming.sources.kafka import KafkaSource
from mage_ai.data_preparation.decorators import streaming_source

@streaming_source
def orders_source(*args, **kwargs) -> KafkaSource:
    return KafkaSource(config=dict(
        bootstrap_servers   = kwargs["context"]["KAFKA_BROKERS"],
        topic               = "orders.events",
        consumer_group      = "mage-orders-5min-count",
        auto_offset_reset   = "latest",
        enable_auto_commit  = False,
        max_poll_records    = 500,
        serde               = "json",
        session_timeout_ms  = 30_000,
    ))

# 2. Tumbling 5-min window transformer — module state + DuckDB
from mage_ai.data_preparation.decorators import transformer
from collections import defaultdict
from datetime import datetime, timedelta
import duckdb, pandas as pd

_buffer: dict[datetime, list[dict]] = defaultdict(list)
WATERMARK_SEC = 60

def _window_start(ts: datetime) -> datetime:
    return ts.replace(second=0, microsecond=0) - timedelta(
        minutes=ts.minute % 5
    )

@transformer
def window_5min_count(batch: list[dict], *args, **kwargs) -> pd.DataFrame:
    now = datetime.utcnow()
    for m in batch:
        ts = datetime.fromisoformat(m["ts"])
        _buffer[_window_start(ts)].append(m)

    watermark = now - timedelta(seconds=WATERMARK_SEC)
    closed = [w for w in list(_buffer.keys()) if w + timedelta(minutes=5) < watermark]
    if not closed:
        return pd.DataFrame()

    rows = []
    for w in closed:
        events_df = pd.DataFrame(_buffer.pop(w))
        con = duckdb.connect(":memory:")
        con.register("e", events_df)
        agg = con.execute("""
            SELECT user_id, COUNT(*) AS order_count, SUM(total_cents)/100.0 AS revenue_usd
            FROM   e GROUP BY user_id
        """).df()
        agg["window_start"] = w
        agg["window_end"]   = w + timedelta(minutes=5)
        rows.append(agg)
    return pd.concat(rows, ignore_index=True)

# 3. Snowflake idempotent MERGE sink
from mage_ai.data_preparation.decorators import streaming_sink
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas
import pandas as pd

@streaming_sink
def snowflake_merge(df: pd.DataFrame, *args, **kwargs) -> None:
    if df.empty:
        return
    conn = snowflake.connector.connect(**kwargs["context"]["SNOWFLAKE_CONN"])
    cur = conn.cursor()
    cur.execute("CREATE OR REPLACE TEMPORARY TABLE stg LIKE analytics.orders_5min")
    write_pandas(conn, df, "STG", schema=None, overwrite=True)
    cur.execute("""
        MERGE INTO analytics.orders_5min tgt
        USING stg src
          ON tgt.user_id = src.user_id AND tgt.window_start = src.window_start
        WHEN MATCHED THEN UPDATE SET
            order_count = src.order_count,
            revenue_usd = src.revenue_usd
        WHEN NOT MATCHED THEN INSERT (user_id, order_count, revenue_usd, window_start, window_end)
                          VALUES (src.user_id, src.order_count, src.revenue_usd, src.window_start, src.window_end)
    """)
    conn.commit()
    conn.close()

Step-by-step trace.

Layer	Config	Reasoning
Kafka source	500 msg batch, no auto-commit	Mage owns offset commit
Buffer	module-level dict per window	in-memory state; bounded
Watermark	60 s	absorb late events
Aggregation	DuckDB SQL per closed window	fast; SQL power
Sink	Snowflake MERGE on (user_id, window_start)	idempotent
Offset commit	after sink returns success	at-least-once + idempotent = effectively exactly-once

After deployment, the pipeline processes ~30k events/sec on a single Mage worker (well under the 100k ceiling), each 5-min window emits ~50k user rows into Snowflake within 60-90 s of window close, and a 10-minute Kafka broker outage causes the pipeline to pause polling (no throughput, no data loss); on broker recovery, Kafka replays from the last committed offset and the pipeline drains the backlog. Snowflake sees no duplicates because the MERGE key is (user_id, window_start).

Output:

Metric	Value
Throughput	~30k events/sec (single worker)
p99 latency (event → Snowflake row)	~90 s
Exactly-once posture	effective (idempotent MERGE + post-sink commit)
State size	~200 MB (bounded by open windows × users)
Kafka outage tolerance	full (offsets replay on recovery)
Failure semantics per batch	retry with same offset; no data loss

Why this works — concept by concept:

enable_auto_commit=false + post-sink commit — the exactly-once cornerstone. Mage never advances the Kafka offset until the Snowflake MERGE has returned success. Failure at any earlier step re-polls the same batch.
Idempotent MERGE — the sink-side half of exactly-once. Because MERGE is deterministic and keyed on (user_id, window_start), replaying the same batch produces the same table state. Combined with post-sink commit, this yields exactly-once effect without a distributed transaction.
Watermark 60 s — the late-event story. Windows close only after their end + 60 s < now, giving late events 60 s to arrive. Longer = more forgiving; shorter = tighter latency. 60 s is the sweet spot for well-behaved sources.
Module-level buffer — the state store. Because Mage keeps the transformer module loaded across batches, module state persists trivially. Bounded by open windows × users; typically under 1 GB for well-behaved streams.
Cost — one Mage worker (2 vCPU, 4 GB RAM), 30k events/sec, ~90 s window latency, Snowflake credit for the MERGE (~$5/day at this rate). Compared to Flink (a separate cluster with JobManager + TaskManagers + RocksDB state backend + savepoint operations), Mage's cost is a fraction and the operational surface is a fraction. The trade is: this scales to ~100k/sec; above that, Flink earns its keep. O(batch size) per iteration; O(open windows × users) memory.

Streaming
Topic — streaming
Streaming windowed aggregation problems

Practice →

Aggregation Topic — aggregation Aggregation problems for real-time windows

Practice →

4. Integrations + destinations

90+ pre-built connectors, first-class dbt, custom Python sources — the integration layer that folds Airbyte's catalogue into the block DAG

The mental model in one line: Mage bundles 90+ pre-built connectors (Postgres, MySQL, SQL Server, Snowflake, BigQuery, Redshift, S3, GCS, Azure Blob, Salesforce, Stripe, GitHub, HubSpot, MongoDB, DynamoDB, Elasticsearch, and more), a first-class dbt block type that runs dbt models as part of a Mage DAG with per-model status tracking, custom Python source and sink support that lets you write your own connector in ~30 lines, secrets management wired to environment variables / HashiCorp Vault / cloud KMS, and — since Airbyte acquired Mage in 2024 — a roadmap that pulls the ~350-connector Airbyte catalogue into the Mage block model over time — all invocable from any block in any pipeline via a one-line configuration. Every integration-heavy interview probe converges on this axis because it is where Mage most obviously wins for teams whose primary pain is "how many connectors do we have to write?"

The pre-built connector catalogue — what's in the box.

Databases. Postgres, MySQL, SQL Server, Oracle, Snowflake, BigQuery, Redshift, Redshift Serverless, ClickHouse, DuckDB, MongoDB, DynamoDB, Elasticsearch. Full read/write with connection-pool management.
Object storage. S3, GCS, Azure Blob, MinIO. Read/write for CSV, Parquet, JSON, Avro, ORC. Partition-aware paths.
SaaS APIs. Salesforce, HubSpot, Stripe, GitHub, Google Analytics, Google Sheets, Intercom, Zendesk, Mixpanel, Segment. OAuth or token-based auth.
Message queues. Kafka, Kinesis, RabbitMQ, Amazon SQS, Google Pub/Sub. For streaming pipelines primarily.
Import mechanism. from mage_ai.io.postgres import Postgres (or Snowflake, S3, Salesforce, etc.). Every connector follows the same with_config(...) → .load(...) / .export(...) interface.

The dbt block — dbt as a first-class Mage citizen.

The type. type: dbt in metadata.yaml. Not a Python block calling subprocess.run(['dbt', ...]) — a first-class block type with its own YAML configuration schema and its own status reporting.
The config. dbt_project_path (path to dbt_project.yml), command (run, test, snapshot, build, seed), select (dbt model selector — tag:daily, +my_model, state:modified), variables (Jinja vars).
The status. Mage's UI shows per-model status (each dbt model appears as a sub-status inside the block), not just "block succeeded / failed." This is qualitatively better than the Airflow BashOperator equivalent.
The typical shape. Upstream Mage blocks stage data into the warehouse; a dbt block runs downstream models tagged appropriately; further Mage blocks trigger downstream side effects (reverse-ETL, notifications).

Custom Python source / sink — writing your own connector.

When to write one. The catalogue covers ~95% of common sources. For the 5% (a proprietary API, an internal message bus, a legacy fixed-width file format), write a custom @data_loader or @data_exporter block using standard Python HTTP / DB libraries.
The shape. ~30 lines: instantiate the client, authenticate, iterate over the source's pagination, yield rows, return a DataFrame. Same signature as any other block — no special "custom connector" ceremony.
Sharing. Custom sources live in the pipeline directory (or a shared custom_sources/ module) and can be reused across pipelines. Mage does not (yet) have a public custom-connector marketplace, but the block-file portability makes copy-paste sharing effortless.

Secrets management — the platform layer, not per-block.

The problem. Every connector needs credentials — Postgres passwords, S3 keys, Salesforce tokens, Snowflake private keys. Hard-coding is unacceptable; scattering os.getenv calls across blocks is fragile.
The Mage answer. A single io_config.yaml file per project defines named "profiles" — each profile is a connection config bundle. Blocks reference profile names, not raw credentials.
Profile sources. Environment variables, HashiCorp Vault, AWS Secrets Manager, GCP Secret Manager. Mage resolves the actual secret at runtime; the profile YAML holds references, not secrets.
Rotation. Rotating a secret means updating the vault; no Mage config change needed. Restart-free (Mage re-reads on each block execution).

Scheduling and triggers.

Cron schedule. Standard cron expressions (0 * * * * hourly, 0 0 * * * daily) attached per pipeline in the UI.
Event triggers. File-arrival on S3 (via a sensor block or an event trigger), webhook (Mage exposes an HTTP endpoint), API call (POST to /api/pipeline_schedules/N/pipeline_runs triggers a run).
Sensor blocks. A @sensor block polls an external condition (file exists, row count exceeds threshold, upstream pipeline completed); downstream blocks unblock when the sensor returns True.
Backfill. UI-driven — pick a date range, kick off runs for each date. Mage handles the parallelism (bounded by a max_concurrent_runs setting per pipeline).

Common interview probes on integrations.

"How does Mage handle secrets?" — required answer: io_config.yaml profiles referencing env / Vault / cloud KMS.
"How does the dbt block compare to Airflow's BashOperator calling dbt?" — first-class per-model status, native block type, not a shell wrapper.
"How do you write a custom connector?" — Python @data_loader / @data_exporter in ~30 lines; no special API.
"What changed after Airbyte acquired Mage?" — 350+ Airbyte connectors converging into the Mage block model over 2024-2026; unified operational plane.

Worked example — Salesforce → Snowflake with a dbt transformation stage

Detailed explanation. The canonical integration-heavy Mage pipeline: pull contacts from Salesforce, stage them to Snowflake, run a dbt model that computes a dim_contacts SCD-2, then trigger a reverse-ETL export of the enriched contacts back to Salesforce as a custom field. Walk through the four blocks and the metadata.yaml.

Source. Salesforce (Contact object, last 24h modified).
Stage. Snowflake STAGING.contacts_raw.
Transform. dbt dim_contacts model (SCD-2 in dbt).
Reverse-ETL. Push enriched attribute back to Salesforce as a custom field.

Question. Write the four blocks and the metadata.yaml.

Input.

Component	Purpose
load_salesforce_contacts	Salesforce Contact SOQL query
stage_to_snowflake	write to STAGING.contacts_raw
dbt_dim_contacts	dbt block, `select: dim_contacts`
push_enriched_to_salesforce	reverse-ETL back to Salesforce

Code.

# File: pipelines/contacts_360/data_loaders/load_salesforce_contacts.py
from mage_ai.data_preparation.decorators import data_loader
from mage_ai.io.salesforce import Salesforce
import pandas as pd

@data_loader
def load_salesforce_contacts(*args, **kwargs) -> pd.DataFrame:
    """Pull Salesforce Contact rows modified in the last 24 hours."""
    sf = Salesforce.with_config(kwargs["context"]["SALESFORCE_PROFILE"])
    soql = (
        "SELECT Id, Email, FirstName, LastName, AccountId, "
        "       LeadSource, LastModifiedDate "
        "FROM   Contact "
        "WHERE  LastModifiedDate > LAST_N_DAYS:1"
    )
    return sf.query(soql)

# File: pipelines/contacts_360/transformers/stage_to_snowflake.py
from mage_ai.data_preparation.decorators import transformer
from mage_ai.io.snowflake import Snowflake
import pandas as pd

@transformer
def stage_to_snowflake(df: pd.DataFrame, *args, **kwargs) -> None:
    """Write to STAGING.contacts_raw — overwrite, atomic replace."""
    sf_writer = Snowflake.with_config(kwargs["context"]["SNOWFLAKE_PROFILE"])
    sf_writer.export(
        df,
        table_name="contacts_raw",
        schema="STAGING",
        database="ACME",
        if_exists="replace",
    )

# The dbt block is pure YAML — no Python file needed
# (metadata.yaml excerpt)
  - uuid: dbt_dim_contacts
    type: dbt
    configuration:
      dbt_project_path: /home/mage/dbt/acme
      command: build
      select: +dim_contacts
      variables:
        run_date: "{{ execution_date.strftime('%Y-%m-%d') }}"
    upstream_blocks:
      - stage_to_snowflake

# File: pipelines/contacts_360/data_exporters/push_enriched_to_salesforce.py
from mage_ai.data_preparation.decorators import data_exporter
from mage_ai.io.salesforce import Salesforce
from mage_ai.io.snowflake import Snowflake
import pandas as pd

@data_exporter
def push_enriched(*args, **kwargs) -> None:
    """Pull enriched dim_contacts from Snowflake; upsert enriched fields back to Salesforce."""
    sf_reader   = Snowflake.with_config(kwargs["context"]["SNOWFLAKE_PROFILE"])
    enriched: pd.DataFrame = sf_reader.load(query=(
        "SELECT contact_id AS Id, lifetime_orders__c, ltv_usd__c "
        "FROM   analytics.dim_contacts "
        "WHERE  updated_at > CURRENT_DATE - 1"
    ))
    if enriched.empty:
        return

    sf_writer = Salesforce.with_config(kwargs["context"]["SALESFORCE_PROFILE"])
    # Bulk upsert 200 at a time (Salesforce API limit)
    for chunk in [enriched[i:i+200] for i in range(0, len(enriched), 200)]:
        sf_writer.bulk_upsert(
            object_type="Contact",
            external_id_field="Id",
            records=chunk.to_dict(orient="records"),
        )

# File: pipelines/contacts_360/metadata.yaml
name: contacts_360
uuid: contacts_360
type: python
blocks:
  - uuid: load_salesforce_contacts
    type: data_loader
    language: python
    upstream_blocks: []
  - uuid: stage_to_snowflake
    type: transformer
    language: python
    upstream_blocks:
      - load_salesforce_contacts
  - uuid: dbt_dim_contacts
    type: dbt
    configuration:
      dbt_project_path: /home/mage/dbt/acme
      command: build
      select: +dim_contacts
    upstream_blocks:
      - stage_to_snowflake
  - uuid: push_enriched_to_salesforce
    type: data_exporter
    language: python
    upstream_blocks:
      - dbt_dim_contacts

Step-by-step explanation.

Salesforce.with_config(...) reads the profile from io_config.yaml — the profile YAML holds a reference to a Vault secret; the actual OAuth token is fetched at runtime. No credentials in Git.
The stager overwrites STAGING.contacts_raw atomically (if_exists="replace"). Because dbt's dim_contacts model reads from the staging table, a partial staging failure leaves the previous staging state intact — dbt runs against a consistent snapshot.
The dbt block is pure configuration — no Python file. command: build runs dbt build, which is run + test in one command; select: +dim_contacts selects dim_contacts and all upstream dependencies. run_date variable is Jinja-templated from Mage's execution_date.
The reverse-ETL block reads from the warehouse (not from the upstream Mage block) because the dbt run is what mutated the warehouse state; the enriched columns don't flow through the block DAG. This is a common Mage pattern for dbt-heavy pipelines.
The Salesforce bulk_upsert in chunks of 200 respects the Salesforce Bulk API row-per-request limit. external_id_field="Id" upserts on Salesforce's own record ID — safe idempotent update.

Output.

Block	Rows	Runtime	Side effect
load_salesforce_contacts	5,000 modified	3 s	(in-memory DataFrame)
stage_to_snowflake	5,000	4 s	STAGING.contacts_raw replaced
dbt_dim_contacts	(dbt)	45 s	analytics.dim_contacts refreshed
push_enriched_to_salesforce	5,000	12 s	Salesforce Contact custom fields updated

Rule of thumb. For any dbt-in-Mage pipeline, use dbt build (not dbt run + dbt test as separate blocks) so that model + test failures halt the DAG in one place. For reverse-ETL, read from the warehouse post-dbt, not from the upstream Mage block — the dbt run is the authoritative refresher.

Worked example — custom Python source for a proprietary internal API

Detailed explanation. The 5% of workloads where Mage's catalogue doesn't cover a source: an internal HTTP API that returns paginated JSON. Write a custom @data_loader in ~30 lines that handles pagination, retries, and rate limiting. This is the "write-your-own-connector in one file" story.

API. https://internal-api.acme.com/v1/orders — paginated, page + page_size query params, returns {data: [...], next: 'page=2'}.
Auth. Bearer token in header.
Rate limit. 100 req/min.
Retry. On 429 (rate limited), wait and retry; on 5xx, exponential backoff.

Question. Write the custom @data_loader block.

Input.

Component	Value
API endpoint	`https://internal-api.acme.com/v1/orders`
Auth	Bearer token from Vault
Pagination	`page` + `page_size=500`
Rate limit	100 req/min
Retry	429 → wait 60s; 5xx → exp backoff

Code.

# File: pipelines/orders_from_internal_api/data_loaders/load_orders_api.py
from mage_ai.data_preparation.decorators import data_loader
import pandas as pd
import requests
import time
from typing import Iterator

@data_loader
def load_orders_api(*args, **kwargs) -> pd.DataFrame:
    """Custom connector for the internal orders API — paginated, retrying."""
    base_url = "https://internal-api.acme.com/v1/orders"
    token    = kwargs["context"]["INTERNAL_API_TOKEN"]     # from Vault via io_config
    headers  = {"Authorization": f"Bearer {token}"}
    all_rows: list[dict] = []
    page = 1

    while True:
        params = {"page": page, "page_size": 500}
        resp = _get_with_retries(base_url, headers=headers, params=params)
        payload = resp.json()
        rows    = payload.get("data", [])
        all_rows.extend(rows)

        if not payload.get("next"):
            break
        page += 1

    df = pd.DataFrame(all_rows)
    print(f"Fetched {len(df)} orders across {page} pages")
    return df

def _get_with_retries(url: str, *, headers: dict, params: dict,
                      max_attempts: int = 5) -> requests.Response:
    """GET with 429 + 5xx retry logic."""
    for attempt in range(1, max_attempts + 1):
        r = requests.get(url, headers=headers, params=params, timeout=30)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", 60))
            print(f"[attempt {attempt}] 429 rate-limited; sleeping {wait}s")
            time.sleep(wait)
            continue
        if 500 <= r.status_code < 600:
            wait = 2 ** attempt
            print(f"[attempt {attempt}] {r.status_code}; backoff {wait}s")
            time.sleep(wait)
            continue
        r.raise_for_status()
        return r
    raise RuntimeError(f"exceeded {max_attempts} attempts against {url}")

Step-by-step explanation.

INTERNAL_API_TOKEN is resolved from the io_config.yaml profile at runtime — the actual token lives in Vault; the profile YAML holds a Vault reference. Zero credentials in Git.
Pagination is a simple while True loop that increments page until the response's next field is empty. Every API's pagination shape is different; this pattern handles the common "cursor in the response" style.
_get_with_retries handles both 429 (rate limited — respect the Retry-After header) and 5xx (server error — exponential backoff). This is the minimum viable retry policy for any HTTP source; more sophisticated implementations would use tenacity or a similar library.
The whole block is ~40 lines including retry logic. There's no Mage-specific ceremony — this is just a @data_loader-decorated function that returns a DataFrame. The same code pattern serves any paginated HTTP API.
To make the connector reusable across pipelines, extract the while loop and _get_with_retries into a shared custom_sources/internal_api.py module and import it from each pipeline's data loader. Mage will resolve the import from the pipeline's Python path.

Output.

Iteration	Page	Rows fetched	Cumulative	Notes
1	1	500	500	200 OK
2	2	500	1,000	200 OK
3	3	(429)	1,000	sleep 60s
3 (retry)	3	500	1,500	200 OK
4	4	234	1,734	last page (no next)

Rule of thumb. For any custom Mage connector, extract the reusable HTTP pagination + retry logic into a shared module and keep the block file focused on the source-specific query. For sources you'll use in more than three pipelines, promote the shared module to a proper Python package and publish it to your internal PyPI.

Worked example — the `io_config.yaml` secrets pattern

Detailed explanation. The Mage answer to "where do credentials live?" is the io_config.yaml file — one file per project that defines named profiles, each profile referencing a vault or environment variable rather than holding the raw secret. Walk through configuring three profiles (Postgres, Salesforce, Snowflake) with three different secret backends.

Profile 1. Postgres — plain env-var reference (dev environment).
Profile 2. Salesforce — HashiCorp Vault reference (prod environment).
Profile 3. Snowflake — AWS Secrets Manager reference (prod environment).

Question. Write the io_config.yaml with three profiles and show how a block references each.

Input.

Profile	Secret backend	Runtime resolution
postgres_dev	env var	`os.getenv`
salesforce_prod	Vault	Vault agent sidecar
snowflake_prod	AWS Secrets Manager	boto3 SecretsManager client

Code.

# File: io_config.yaml — project-level secrets manifest
version: 0.1.1
default:
  # Postgres — dev, env-var backed
  POSTGRES_CONNECT_TIMEOUT: 10
  POSTGRES_DBNAME: acme
  POSTGRES_SCHEMA: public
  POSTGRES_USER: acme_dev
  POSTGRES_PASSWORD: "{{ env_var('POSTGRES_PASSWORD') }}"
  POSTGRES_HOST: db-dev.internal
  POSTGRES_PORT: 5432

  # Salesforce — prod, Vault backed
  SALESFORCE_DOMAIN: acme.my.salesforce.com
  SALESFORCE_USERNAME: "{{ vault('secret/salesforce', 'username') }}"
  SALESFORCE_PASSWORD: "{{ vault('secret/salesforce', 'password') }}"
  SALESFORCE_SECURITY_TOKEN: "{{ vault('secret/salesforce', 'security_token') }}"
  SALESFORCE_CLIENT_ID: "{{ vault('secret/salesforce', 'client_id') }}"
  SALESFORCE_CLIENT_SECRET: "{{ vault('secret/salesforce', 'client_secret') }}"

  # Snowflake — prod, AWS Secrets Manager backed
  SNOWFLAKE_USER: acme_svc
  SNOWFLAKE_ACCOUNT: acme-prod.us-east-1
  SNOWFLAKE_WAREHOUSE: LOAD_WH
  SNOWFLAKE_DATABASE: ACME
  SNOWFLAKE_SCHEMA: PUBLIC
  SNOWFLAKE_PRIVATE_KEY_PASSPHRASE: "{{ aws_secretsmanager('acme/snowflake', 'private_key_passphrase') }}"
  SNOWFLAKE_PRIVATE_KEY: "{{ aws_secretsmanager('acme/snowflake', 'private_key_pem') }}"

# Any block references the profile by name — Mage resolves at runtime
from mage_ai.data_preparation.decorators import data_loader
from mage_ai.io.postgres import Postgres
from mage_ai.data_preparation.shared.secrets import get_secret_value
import pandas as pd

@data_loader
def load_from_postgres(*args, **kwargs) -> pd.DataFrame:
    # The profile 'default' pulls the entire block above; Mage resolves
    # env_var(), vault(), and aws_secretsmanager() at load time.
    pg = Postgres.with_config(get_secret_value("default"))
    return pg.load("SELECT * FROM public.orders LIMIT 100")

Step-by-step explanation.

io_config.yaml is the place secrets live at reference level. The file itself is committed to Git — because it holds only references (env_var('...'), vault('...', '...'), aws_secretsmanager('...', '...')), never raw values. Reviewers can safely see the profile shape.
The three secret backends are pluggable: env_var('...') reads from the environment (development), vault('...', '...') reads from a HashiCorp Vault (Vault agent sidecar authenticated), aws_secretsmanager('...', '...') reads from AWS Secrets Manager (via boto3 with the Mage worker's IAM role).
Rotating a secret is a Vault / AWS action — no Mage config change. The next block execution re-resolves the reference and picks up the new value. This is the correct rotation posture.
Multiple profiles in the same file (e.g., default, prod, staging) are addressed by name; blocks pick one via the profile parameter to with_config. This lets a single Mage instance run pipelines against multiple environments without file swapping.
Never write a block that constructs a connection string from raw kwargs — that pattern leaks secrets to logs and re-implements what io_config already does correctly. Always go through with_config(profile_name) for consistency.

Output.

Profile	Backend	Raw value in Git?	Rotation
postgres_dev	env var	no (env only)	env-var re-set + restart
salesforce_prod	Vault	no (Vault ref)	Vault write; auto-refresh
snowflake_prod	AWS SM	no (AWS SM ref)	AWS SM write; auto-refresh

Rule of thumb. Every production Mage deployment must have io_config.yaml reference-only (never raw values) and every block must load credentials via with_config(profile_name) — this keeps secrets out of Git, out of logs, and out of block code. Rotating secrets is a vault operation, not a deploy.

Senior interview question on Mage integrations

A senior interviewer might ask: "Your team uses Airbyte for ~40 SaaS ingest jobs, dbt for ~200 models, and Airflow to orchestrate them. You're evaluating consolidating on Mage. Design the migration: which workloads move first, how the dbt integration changes, how Airbyte connectors continue to work, and the operational model post-Airbyte-acquisition of Mage."

Solution Using dbt block for models + Airbyte-connector-backed loaders + a phased migration anchored on the acquisition roadmap

# 1. Migration plan — three-phase, workload-scored
phase_1_dbt_wrap:
  duration: 3 weeks
  action: wrap every existing dbt run in a Mage dbt block
  benefit: unified UI for dbt + downstream; per-model status in Mage UI
  risk: near-zero (dbt itself unchanged)

phase_2_airbyte_backed_loaders:
  duration: 6 weeks
  action: convert Airbyte connections into Mage data-loader blocks
    using the Airbyte-Mage bridge (post-acquisition)
    OR retain Airbyte for the SaaS pulls, add Mage sensor blocks
    that wait for Airbyte completion
  benefit: single scheduling plane; Airflow's Airbyte operators retired
  risk: low (connectors themselves unchanged)

phase_3_airflow_decommission:
  duration: 4 weeks
  action: retire Airflow scheduler + web UI; keep MWAA cluster paused for 30d
  benefit: one orchestration surface; ~40% ops overhead reduction
  risk: managed — Airflow un-pauseable during 30d window

# 2. Mage pipeline: Airbyte-sensor + dbt block + reverse-ETL
name: daily_finance_close
uuid: daily_finance_close
type: python

blocks:
  # Sensor waits for the Airbyte Stripe connection to succeed
  - uuid: wait_airbyte_stripe
    type: sensor
    language: python
    configuration:
      # Polls Airbyte API for the connection's last successful sync
      airbyte_connection_id: "conn_stripe_daily"
      poll_interval_sec: 30
      timeout_sec: 3600
    upstream_blocks: []

  # Sensor waits for the Airbyte Salesforce connection to succeed
  - uuid: wait_airbyte_salesforce
    type: sensor
    language: python
    configuration:
      airbyte_connection_id: "conn_salesforce_daily"
      poll_interval_sec: 30
      timeout_sec: 3600
    upstream_blocks: []

  # dbt run once both sources have landed
  - uuid: dbt_finance_close
    type: dbt
    configuration:
      dbt_project_path: /home/mage/dbt/acme
      command: build
      select: +fct_finance_close
    upstream_blocks:
      - wait_airbyte_stripe
      - wait_airbyte_salesforce

  # Reverse-ETL: push closed-day metrics to Slack
  - uuid: post_daily_close_to_slack
    type: data_exporter
    language: python
    upstream_blocks:
      - dbt_finance_close

# 3. The Airbyte-sensor block (simplified)
from mage_ai.data_preparation.decorators import sensor
import requests
import time

@sensor
def wait_airbyte(*args, **kwargs) -> bool:
    """Poll Airbyte API; return True when the connection's last sync succeeded today."""
    conn_id = kwargs["configuration"]["airbyte_connection_id"]
    api     = kwargs["context"]["AIRBYTE_API_URL"]
    hdr     = {"Authorization": f"Bearer {kwargs['context']['AIRBYTE_TOKEN']}"}
    resp = requests.get(f"{api}/v1/connections/{conn_id}", headers=hdr)
    latest = resp.json().get("latestSyncJobStatus")
    return latest == "succeeded"

Step-by-step trace.

Phase	Action	Benefit	Duration
1 (dbt wrap)	Every dbt run → Mage dbt block	unified UI, per-model status	3 weeks
2 (Airbyte bridge)	Sensor blocks OR Mage-native loaders	single scheduling plane	6 weeks
3 (Airflow decomm)	Retire Airflow scheduler	~40% ops reduction	4 weeks
Ongoing	Airbyte acquisition roadmap	~350 connectors converge into Mage	2024→2026

After the phased migration, the team's daily finance close runs entirely in Mage: two sensor blocks wait for the Airbyte Stripe and Salesforce syncs, one dbt block builds fct_finance_close, one exporter posts the summary to Slack. Airflow is decommissioned. The Airbyte acquisition roadmap means that over 2025-2026, the sensor pattern is replaced by native Mage loaders that call Airbyte connectors directly — unified operational plane, single scheduler.

Output:

Component	Before	After
Orchestrator	Airflow	Mage
dbt runs	BashOperator	Mage dbt block (native)
Airbyte handoff	AirbyteTriggerSyncOperator	Mage sensor OR native loader
Per-model status	none	Mage UI
Ops surface	2 (Airflow + Airbyte)	1 (Mage, post-roadmap)
Migration risk	phased, per-workload	low (each phase reversible)

Why this works — concept by concept:

Phase 1 dbt wrap — wrapping dbt in a Mage block requires zero dbt project changes. The dbt CLI still runs; Mage only adds the scheduling + UI layer. Near-zero migration risk.
Phase 2 Airbyte bridge — sensor blocks poll Airbyte's connection status; downstream Mage blocks trigger on success. This decouples Airbyte's connector runs from the Mage DAG while keeping both operational. Post-acquisition, sensors are replaced by native loaders.
Phase 3 Airflow decommission — only after phases 1 and 2 are proven for 30 days. Keeps the MWAA cluster paused (not deleted) for rollback safety.
Airbyte acquisition roadmap — the 2024 acquisition means Mage and Airbyte converge over time. Betting on Mage now positions the team for the unified plane; running Airbyte + Mage in parallel today is not wasted work — the sensor pattern retires gracefully.
Cost — 13 engineer-weeks of migration work over 3 months, offset by ~40% ongoing ops overhead reduction (single orchestrator, single UI, single scheduler). Payback in ~6-9 months. Net-positive by year 1; qualitatively better UX for engineers throughout.

Design
Topic — design
Design problems on integration and reverse-ETL pipelines

Practice →

SQL Topic — sql SQL warehouse-side transformation problems

Practice →

5. Mage vs Dagster / Prefect / Airflow + interview signals

The four-tool decision matrix — Mage wins hands-on UX, Dagster wins SDA lineage, Prefect wins dynamic flows, Airflow wins legacy scale

The mental model in one line: mage ai vs dagster is the interview shorthand for the broader four-tool comparison — Mage AI, Dagster, Prefect, and Apache Airflow all schedule DAGs but differentiate on four axes (dev-loop UX, SDA lineage granularity, dynamic Python flow support, and streaming), and the correct 2026 pick is workload-dependent rather than tool-dependent — Mage wins for teams whose primary constraint is iteration velocity and mixed batch + streaming, Dagster wins for regulated analytics-engineering where asset lineage is audit-critical, Prefect wins for dynamic per-tenant flow shapes, Airflow wins for legacy DAG scale where migration cost dominates any UX win — and senior interviewers probe whether you can articulate the trade-off rather than defend a favourite. Getting this comparison right is the load-bearing question in every senior orchestration interview in 2026.

Axis 1 — Dev-loop UX.

Mage. Browser-native notebook IDE; live preview of each block's output; author-to-first-run in < 5 minutes. The clear category winner. Every block is a file; every file is a decorated function; the DAG is the wiring graph.
Dagster. UI-driven materialisation from the asset graph; author-to-first-run in ~6 minutes. Better than Airflow, worse than Mage.
Prefect. Python-first; @flow / @task decorators; author in an editor, run via python flow.py; author-to-first-run in ~5 minutes for simple flows.
Airflow. DAG files with PythonOperator boilerplate; airflow tasks test for local iteration; author-to-first-run in ~9 minutes. The clear loser on this axis, though the TaskFlow API (2.4+) narrows the gap.

Axis 2 — Software-defined asset (SDA) lineage.

Dagster. Asset graph is the primary abstraction. Every asset knows its dependencies, its materialisation history, its data quality checks. Best in category for lineage. If "which asset produced this row?" matters, Dagster.
Mage. Block-level DAG only. Lineage is per-block, not per-asset. Sufficient for "block A ran before block B" audits; insufficient for "which model produced customer 42's LTV" traces.
Prefect. Flow-run history only. No first-class asset concept. Weak for lineage.
Airflow. No native lineage. AIRFLOW__CORE__LINEAGE_BACKEND hooks OpenLineage, but it's bolted on.

Axis 3 — Dynamic Python flows.

Prefect. Best-in-class. .map() on tasks, dynamic subflows, per-run flow-shape decisions. If your DAG changes shape per run (per-tenant, per-file, per-config), Prefect.
Airflow. Dynamic DAGs (2.3+) via .expand() are decent but less ergonomic than Prefect's mapping.
Dagster. Dynamic partitions + dynamic outs — workable but not the primary paradigm.
Mage. Weakest here. The block DAG is essentially static; dynamic behaviour lives inside a single block, not across blocks. If per-tenant DAG shape is critical, Mage is the wrong pick.

Axis 4 — Streaming.

Mage. First-class streaming pipeline mode with Kafka / Kinesis / RabbitMQ / SQS / Pub/Sub sources, in-block DuckDB / Materialize / RisingWave engines, post-sink offset commit. Category winner.
Dagster. Sensors + partitioned assets can simulate streaming for periodic materialisation, but no native continuous execution.
Prefect. Workers can run long-lived flows; streaming is possible but not idiomatic.
Airflow. KafkaOperator exists but the scheduler-first model is a poor fit for continuous consumption; usually paired with a separate streaming platform.

Axis 5 — Community + ecosystem scale.

Airflow. Largest community by far. Thousands of production deployments. Managed offerings from AWS MWAA, GCP Composer, Astronomer. Every operator you'd want already exists.
Dagster. Growing fast; healthy contributor base; Dagster Cloud managed.
Prefect. Similar scale to Dagster; Prefect Cloud managed.
Mage. Smallest of the four but growing; Airbyte acquisition boost; Mage Pro managed.

The migration paths — which direction each tool bleeds toward Mage.

Airflow → Mage. Lift PythonOperator tasks 1:1 into @data_loader / @transformer / @data_exporter blocks. Keep DAG(...) definition as a scratchpad reference until Mage metadata.yaml is battle-tested. Migration cost ~2 weeks per 10 DAGs.
Prefect → Mage. Port @flow / @task to Mage blocks; the paradigms are similar. Lose dynamic mapping. Migration cost ~1 week per 10 flows.
dbt-only → Mage. Add loader + exporter blocks around your existing dbt project via the dbt block; near-zero disruption. Migration cost ~1 week total.
Dagster → Mage. Rare in practice — teams on Dagster picked it for SDA lineage; migrating away means giving that up. Not recommended unless the lineage requirement disappeared.

Interview signals — the questions senior interviewers ask.

"Compare Mage and Dagster on lineage." — required answer: Dagster's asset graph wins; Mage has block-level only.
"When would you pick Mage over Airflow?" — dev-loop speed + streaming + integrations; not "when your Airflow is old."
"What did the Airbyte acquisition change for Mage?" — connector-catalogue convergence; unified operational plane.
"Describe the Mage streaming pipeline shape." — Kafka source → transformer → sink → post-sink offset commit.
"What's Mage's weakest axis?" — dynamic Python flows (Prefect wins there); SDA lineage (Dagster wins).

Worked example — the four-axis comparison whiteboard

Detailed explanation. Every senior orchestrator interview lands on a whiteboard comparison. The interviewer will ask you to "draw the matrix" for Mage vs Dagster vs Prefect vs Airflow on four axes; the answer that scores highest is a memorised 4x4 table plus one-sentence justifications per cell. Walk through the canonical whiteboard.

Axis 1. Dev-loop UX.
Axis 2. SDA lineage.
Axis 3. Dynamic flows.
Axis 4. Streaming.

Question. Draw the 4x4 matrix and rate each tool 1-5 on each axis, with a one-sentence justification per cell.

Input.

Axis	Mage	Dagster	Prefect	Airflow
Dev-loop UX	5 (live preview in browser)	4 (UI materialise)	4 (Python + editor)	2 (DAG boilerplate)
SDA lineage	2 (block-level only)	5 (asset graph)	2 (flow-run only)	1 (bolted on)
Dynamic flows	2 (static block DAG)	3 (dynamic partitions)	5 (`.map()` + subflows)	3 (2.3+ dynamic DAGs)
Streaming	5 (first-class)	2 (sensor-based)	3 (long-lived workers)	2 (KafkaOperator retrofit)

Code.

# The one-page "which tool wins which axis" table
TOOLS = ["Mage", "Dagster", "Prefect", "Airflow"]
AXES  = ["Dev-loop UX", "SDA lineage", "Dynamic flows", "Streaming"]
RATINGS = {
    "Mage":    [5, 2, 2, 5],
    "Dagster": [4, 5, 3, 2],
    "Prefect": [4, 2, 5, 3],
    "Airflow": [2, 1, 3, 2],
}

print(f"{'axis':<20}", *[f"{t:>10}" for t in TOOLS])
for i, ax in enumerate(AXES):
    print(f"{ax:<20}", *[f"{RATINGS[t][i]:>10}" for t in TOOLS])

# Which axis wins which tool?
for i, ax in enumerate(AXES):
    winner = max(TOOLS, key=lambda t: RATINGS[t][i])
    print(f"{ax:<20} → {winner}")

Step-by-step explanation.

Dev-loop UX: Mage wins 5-4-4-2. The live-preview loop is a category-differentiating feature. Prefect and Dagster are close; Airflow is the clear loser.
SDA lineage: Dagster wins 5-2-2-1. The asset graph is Dagster's identity; the other three all lose. Mage's block-level lineage is a fallback, not an answer.
Dynamic flows: Prefect wins 5-3-3-2. .map() is the killer feature; nothing else matches its ergonomics. Airflow's dynamic DAGs (2.3+) are a workable second-best.
Streaming: Mage wins 5-3-2-2. First-class continuous pipelines with post-sink offset commit; the others require workarounds.
The interview inference: each tool has one axis it clearly wins. Pick based on which axis matters most for your workload. There is no universal winner; there is no universal loser.

Output.

Axis	Winner
Dev-loop UX	Mage
SDA lineage	Dagster
Dynamic flows	Prefect
Streaming	Mage
Community scale	Airflow

Rule of thumb. Memorise the 4x5 (four tools, five axes) matrix cold. Every senior orchestrator interview lands on a variant of this whiteboard. The candidate who names a winner per axis without hedging scores highest.

Worked example — the "when Mage is the wrong pick" scenarios

Detailed explanation. Every strong tool has failure modes. Senior architects earn credibility by naming when not to pick their preferred tool. Walk through three canonical scenarios where Mage is genuinely the wrong choice, plus what to pick instead.

Scenario A. SDA-lineage-critical regulated analytics engineering.
Scenario B. Highly dynamic per-tenant DAG shape.
Scenario C. Streaming above the 100k events/sec ceiling.

Question. For each scenario, name the correct tool and the diagnostic question that reveals the mismatch.

Input.

Scenario	Symptom	Correct tool	Diagnostic
Regulated analytics	audit needs per-asset lineage	Dagster	"who produced customer 42's LTV?"
Per-tenant DAG shape	each tenant has different steps	Prefect	"does DAG shape depend on config?"
500k events/sec streaming	throughput above single-worker ceiling	Flink	"what's the peak events/sec?"
5000-DAG Airflow shop	migration cost > win	Airflow (stay)	"how many existing DAGs?"

Code.

# Anti-recommendation helper — when to NOT pick Mage
def is_mage_wrong_pick(needs_sda_lineage: bool,
                       dag_shape_dynamic_per_tenant: bool,
                       peak_events_per_sec: int,
                       existing_airflow_dags: int) -> tuple[bool, str]:
    if needs_sda_lineage:
        return True, "Pick Dagster — SDA lineage is the load-bearing feature"
    if dag_shape_dynamic_per_tenant:
        return True, "Pick Prefect — dynamic mapping is the load-bearing feature"
    if peak_events_per_sec > 100_000:
        return True, "Pick Flink — Mage streaming ceiling is ~100k/s"
    if existing_airflow_dags > 500:
        return True, "Stay on Airflow — migration cost dominates win"
    return False, "Mage is a reasonable pick"

print(is_mage_wrong_pick(True,  False, 10_000,  0))
# → (True, 'Pick Dagster — SDA lineage is the load-bearing feature')
print(is_mage_wrong_pick(False, True,  10_000,  0))
# → (True, 'Pick Prefect — dynamic mapping is the load-bearing feature')
print(is_mage_wrong_pick(False, False, 500_000, 0))
# → (True, 'Pick Flink — Mage streaming ceiling is ~100k/s')
print(is_mage_wrong_pick(False, False, 10_000,  1000))
# → (True, 'Stay on Airflow — migration cost dominates win')
print(is_mage_wrong_pick(False, False, 10_000,  0))
# → (False, 'Mage is a reasonable pick')

Step-by-step explanation.

Scenario A — regulated analytics. If an auditor needs to trace "which asset produced customer 42's LTV?", Mage's block-level lineage is insufficient. Dagster's asset graph is the load-bearing feature.
Scenario B — dynamic per-tenant DAG. If tenant A has steps 1-2-3 and tenant B has steps 1-2-4-5, the DAG shape depends on runtime config. Mage's static block DAG makes this awkward; Prefect's .map() + subflow model makes it natural.
Scenario C — 500k events/sec streaming. Above Mage's single-worker ceiling (~100k/s). Flink's cluster architecture with RocksDB state backend and savepoint operations earns its complexity at this throughput.
Scenario D — 5000-DAG Airflow. The migration cost (~250 engineer-weeks at 2 weeks per 40 DAGs) dominates any UX win. Stay on Airflow; improve incrementally with TaskFlow API.
The diagnostic pattern: ask about the load-bearing constraint before naming a tool. Every candidate who reflexively answers "Mage" without probing the constraint scores lower than the candidate who says "depends on your streaming throughput / lineage requirement / DAG shape."

Output.

Scenario	Wrong pick	Correct pick	Diagnostic
Regulated audit	Mage	Dagster	asset lineage requirement
Per-tenant DAG shape	Mage	Prefect	dynamic flow requirement
500k/s streaming	Mage	Flink	throughput ceiling
5000 Airflow DAGs	Mage	Airflow (stay)	migration cost

Rule of thumb. Every senior architect answer to "should we adopt Mage?" starts with three diagnostic questions: (1) is SDA lineage audit-critical? (2) is DAG shape dynamic per tenant? (3) is streaming above 100k/s? Only after those three come back "no" does Mage become the natural recommendation.

Worked example — the five interview questions to rehearse

Detailed explanation. The senior Mage interview probes five predictable questions. Rehearsing crisp answers for each one is the difference between a rambling response and a hire. Walk through the five and the two-sentence answer for each.

Q1. "What's the Mage block DAG model?"
Q2. "How does Mage handle streaming?"
Q3. "Compare Mage and Dagster on lineage."
Q4. "What changed after Airbyte acquired Mage?"
Q5. "When is Mage the wrong pick?"

Question. Draft two-sentence answers for each of the five interview probes.

Input.

Question	Length	Key phrase to include
Q1 block DAG	2 sentences	"five block types + implicit dependencies"
Q2 streaming	2 sentences	"post-sink offset commit + DuckDB in transformer"
Q3 vs Dagster	2 sentences	"block-level vs asset-graph lineage"
Q4 Airbyte	2 sentences	"connector-catalogue convergence + unified plane"
Q5 wrong pick	2 sentences	"SDA lineage / dynamic flow / 100k+ streaming"

Code.

Rehearsed Mage interview answers (two sentences each)
======================================================

Q1: "What's the Mage block DAG model?"
A: "Every Mage pipeline is composed of typed blocks — @data_loader,
    @transformer, @data_exporter, @sensor, @custom — with implicit
    dependencies declared via each block's function signature. The DAG
    is the wiring graph of block files; upstream block outputs are
    passed as positional arguments to downstream blocks, and every
    block previews its output live in the browser IDE while you edit."

Q2: "How does Mage handle streaming?"
A: "Streaming pipelines use type: streaming in metadata.yaml, with a
    Kafka / Kinesis / RabbitMQ / SQS / Pub/Sub source block, a
    transformer that often uses DuckDB / Materialize / RisingWave for
    in-block SQL, and a sink block. The source offset is committed
    ONLY after the sink acknowledges — combined with an idempotent
    sink (Snowflake MERGE, S3 deterministic keys) this yields
    effectively exactly-once semantics under the 100k events/sec
    single-worker ceiling."

Q3: "Compare Mage and Dagster on lineage."
A: "Dagster's software-defined-asset graph is the primary abstraction —
    every asset knows its dependencies, its materialisation history,
    its data quality checks, so 'which asset produced this row?' is
    a first-class query. Mage's DAG is block-level, so 'block A ran
    before block B' is queryable but not 'which model produced
    customer 42's LTV' — for regulated analytics engineering,
    Dagster wins; for iteration velocity, Mage wins."

Q4: "What changed after Airbyte acquired Mage?"
A: "Airbyte acquired Mage in 2024, and the roadmap since has been
    connector-catalogue convergence — Airbyte's ~350 connectors
    landing progressively as native Mage data-loader blocks, plus
    a unified operational plane where Airbyte-managed source runs
    and Mage-managed downstream transforms share one scheduler
    and one UI. This means betting on Mage today positions you for
    the merged product, not against it."

Q5: "When is Mage the wrong pick?"
A: "Three scenarios: (a) SDA-lineage-critical regulated analytics
    engineering — pick Dagster; (b) DAG shape that depends on
    per-tenant runtime config — pick Prefect for its dynamic
    mapping; (c) streaming throughput above ~100k events/sec —
    escalate to Flink for its RocksDB state backend. Plus the
    'legacy Airflow with 500+ DAGs' case where migration cost
    dominates any UX win — stay on Airflow, improve incrementally."

Step-by-step explanation.

Q1 rehearsal names the five block types, the implicit-dependency wiring, and the live-preview loop. Any interviewer scoring the answer will hear the load-bearing keywords ("five block types," "implicit dependencies," "live preview").
Q2 rehearsal names the streaming pipeline type, the source-transformer-sink shape, the DuckDB / Materialize / RisingWave engines, the post-sink offset commit, and the 100k/s ceiling. Five load-bearing phrases in two sentences.
Q3 rehearsal frames the trade-off cleanly: Dagster wins lineage, Mage wins velocity. Naming the "which asset produced customer 42's LTV?" query as the load-bearing lineage question shows depth.
Q4 rehearsal names the year (2024), the direction (connector-catalogue convergence), and the strategic implication (betting on Mage positions for the merged product, not against it). Senior signal.
Q5 rehearsal names three scenarios crisply and adds the "legacy Airflow migration cost" case as bonus depth. This is the honest-architect answer that scores highest.

Output.

Q	Two-sentence answer includes
Q1 block DAG	5 block types + implicit deps + live preview
Q2 streaming	type=streaming + post-sink commit + idempotent sink
Q3 vs Dagster	asset graph vs block-level lineage
Q4 Airbyte	2024 acquisition + connector convergence + unified plane
Q5 wrong pick	3 scenarios + Airflow migration cost caveat

Rule of thumb. Rehearse the five two-sentence answers verbatim before every Mage interview. Naming the load-bearing keywords ("five block types," "post-sink offset commit," "asset graph vs block-level," "connector convergence," "Mage wrong pick when...") is the difference between a fluent answer and a stumbling one.

Senior interview question on Mage vs the field

A senior interviewer might ask: "You're the incoming Head of Data at a Series C startup. They have 30 Airflow DAGs, a small Prefect experiment, no dbt yet, and they're planning to add real-time analytics. Walk me through your orchestration strategy for the next 18 months — what do you keep, what do you migrate, what do you add, and how do you justify each choice to the CTO?"

Solution Using a workload-scored migration plan anchored on Mage for greenfield and Airflow-stay for legacy

# 1. Workload inventory + scoring
WORKLOADS = [
    # (name, tool_today, migration_score, streaming?, lineage_critical?)
    ("customer_daily_etl",       "airflow", "keep",     False, False),  # 20 DAGs; migration too costly
    ("finance_daily_close",      "airflow", "keep",     False, True),   # regulated; keep on Airflow for audit
    ("marketing_experiments",    "prefect", "migrate",  False, False),  # small; consolidate on Mage
    ("realtime_clickstream",     "new",     "add",      True,  False),  # greenfield; Mage streaming
    ("dbt_analytics_new",        "new",     "add",      False, False),  # greenfield; Mage dbt block
    ("per_tenant_onboarding",    "prefect", "keep",     False, False),  # dynamic; Prefect wins
]

for name, tool, action, is_streaming, is_regulated in WORKLOADS:
    reason = "Mage" if action == "add" else \
             ("Airflow (legacy scale)"   if action == "keep" and tool == "airflow" else
              "Prefect (dynamic flows)"  if action == "keep" and tool == "prefect" else
              "Mage (consolidate)")
    print(f"{action:8s} {name:30s} → {reason}")

# 2. 18-month roadmap
month_1_to_3:
  action: pilot Mage on the two greenfield workloads
  workloads:
    - realtime_clickstream        # Mage streaming pilot
    - dbt_analytics_new           # Mage dbt block pilot
  success_criteria:
    - streaming p99 < 30s
    - dbt runs 100% on schedule

month_4_to_6:
  action: consolidate marketing_experiments off Prefect onto Mage
  workloads:
    - marketing_experiments
  benefit: single UI for greenfield + marketing

month_7_to_12:
  action: keep Airflow (30 DAGs) — no migration
  rationale: migration cost > win; TaskFlow API upgrade meets UX complaints
  investment: upgrade Airflow to 2.9+; adopt @task decorator; add OpenLineage
  outcome: two orchestrators — Mage for greenfield + marketing, Airflow for legacy

month_13_to_18:
  action: evaluate Airflow → Mage migration for a subset of workloads
  criteria: workloads that need streaming OR heavy integrations
  outcome: incremental migration OR retain two-orchestrator posture indefinitely

# 3. CTO-facing justification (one page)
JUSTIFICATION = """
Strategy in one sentence:
  Adopt Mage for all greenfield work (streaming, dbt-heavy, integration-heavy);
  keep Airflow for existing 30 DAGs; keep Prefect for the one dynamic
  per-tenant onboarding flow.

Why not consolidate everything on one tool:
  - Airflow migration for 30 DAGs = ~30 engineer-weeks; win is dev-loop UX
    which the TaskFlow API upgrade largely closes anyway.
  - Prefect wins dynamic per-tenant flows; Mage's static block DAG is
    a poor fit for that shape.

Why Mage for greenfield:
  - Streaming is first-class; Airflow needs a separate platform.
  - dbt block is native; Airflow's BashOperator is a workaround.
  - Dev-loop is measurably 2x faster than Airflow (5 min vs 9 min).
  - Airbyte acquisition roadmap positions us for the merged product.

Risk mitigation:
  - Pilot on two workloads for 3 months before broader rollout.
  - Airflow stays paused-not-decommissioned during any migration.
  - Two-orchestrator posture is a supported end-state, not a transition.

Total cost:
  - 3-6 engineer-months of pilot + consolidation.
  - Ongoing: ~40% less orchestrator ops overhead vs an all-Airflow world
    for the greenfield workloads (single UI, native dbt, native streaming).
"""
print(JUSTIFICATION)

Step-by-step trace.

Step	Decision	Reasoning
Score every workload	on migration_score + streaming + lineage	tool-first is wrong; workload-first is right
Keep Airflow (30 DAGs)	too costly to migrate	~30 engineer-weeks for a UX win TaskFlow closes
Keep Prefect (1 flow)	dynamic per-tenant shape	Mage is wrong tool for this
Add Mage for greenfield	streaming + dbt + integrations	tool wins on all three axes
Consolidate marketing	small workload; already growing	single UI benefit
Two-orchestrator end-state	not a transition	honest architect answer

After the 18-month roadmap, the team runs Mage (~8 pipelines including streaming + dbt-heavy analytics), Airflow (30 legacy DAGs, TaskFlow-modernised), and Prefect (1 per-tenant flow). Each tool serves its load-bearing workload; the CTO sees measurable wins (dev-loop speed on greenfield, streaming p99 under 30s, no disruption to regulated finance close) with bounded migration cost.

Output:

Component	Before	After
Orchestrator count	2 (Airflow + Prefect)	3 (+ Mage)
Greenfield tool	Airflow (would-have)	Mage
Streaming platform	none	Mage streaming
dbt integration	none	Mage dbt block
Legacy DAGs	30 on Airflow	30 on Airflow (TaskFlow upgraded)
Migration engineer-weeks	(would have been ~30)	~6 (Mage pilot + Prefect consolidate)
Ops overhead per orchestrator	1x	0.6x for Mage-managed workloads

Why this works — concept by concept:

Workload-first, not tool-first — every workload scored on its constraints, not evaluated against a preferred tool. This is the senior architect posture that scores highest in interviews.
Airflow stays for legacy — migration cost for 30 DAGs is ~30 engineer-weeks. Weighed against a dev-loop UX win, that math loses. TaskFlow API upgrade closes most of the UX gap incrementally.
Mage for greenfield — the three axes Mage wins (dev-loop, streaming, integrations) all matter for greenfield. Pilot small, prove out, expand.
Prefect for dynamic flows — the one workload where Prefect is the load-bearing correct answer. Recognising that and keeping it is more senior than reflexively consolidating.
Cost — ~6 engineer-months of migration + pilot work over 18 months, offset by ~40% ops overhead reduction on greenfield-side, plus a measurably faster dev-loop that compounds across ongoing iteration. Two-orchestrator end-state is a feature, not a bug — right tool for each workload. Payback in ~12 months on the ops side; UX win compounds indefinitely.

Design
Topic — design
Design problems on orchestrator strategy and roadmap

Practice →

Streaming
Topic — streaming
Streaming architecture and Kafka consumer problems

Practice →

Cheat sheet — Mage recipes

Which block when. @data_loader for source ingestion (Postgres, S3, Kafka, custom API); @transformer for shape / join / aggregate (SQL / pandas / DuckDB / PySpark); @data_exporter for sink writes (Snowflake, S3, Salesforce, Kafka); @sensor for external condition blocking (file arrival, upstream pipeline completion, Airbyte connection status); @custom (scratchpad) for one-off explorations that shouldn't be scheduled. Every non-scratchpad block is one function per file; the DAG is the wiring graph declared in metadata.yaml.
Three-block pipeline skeleton. pipelines/<name>/data_loaders/load_X.py with @data_loader def load_X(*args, **kwargs) -> pd.DataFrame; pipelines/<name>/transformers/enrich_X.py with @transformer def enrich(df, *args, **kwargs) -> pd.DataFrame; pipelines/<name>/data_exporters/write_X.py with @data_exporter def write(df, *args, **kwargs) -> None; pipelines/<name>/metadata.yaml listing the three blocks with upstream_blocks: [<uuid>] per block. The subdirectory names must match the block types exactly for Mage's decorator check to pass.
Streaming pipeline skeleton. type: streaming in metadata.yaml; source block returns a KafkaSource(...) object with enable_auto_commit=false and max_poll_records=500; transformer receives list[dict] batches (not a DataFrame); optional in-block DuckDB / Materialize / RisingWave for windowing; sink is idempotent (Snowflake MERGE, S3 deterministic keys, Postgres ON CONFLICT); Mage commits Kafka offsets after the sink returns success. Under ~100k events/sec + ~2 GB state Mage streaming is the right pick; above that ceiling escalate to Flink.
dbt-in-Mage handoff. type: dbt block in metadata.yaml with configuration: {dbt_project_path, command: build, select: +my_model, variables: {...}}. Not a Python file — pure YAML. Runs dbt build --select +my_model under the hood and surfaces per-model status in the Mage UI. Upstream Mage blocks stage data into the warehouse; the dbt block runs models; downstream Mage blocks trigger reverse-ETL from the post-dbt warehouse state. Prefer dbt build over separate dbt run + dbt test so model + test failures halt in one place.
io_config.yaml secrets recipe. One project-level file with named profiles; every profile value is a reference (env_var('...'), vault('secret/path', 'key'), aws_secretsmanager('secret/name', 'key')), never a raw secret. Blocks load credentials via Postgres.with_config(get_secret_value('default')) (or the equivalent for Snowflake, Salesforce, S3). Rotating a secret is a vault operation; no Mage config change or restart needed. Never construct connection strings from raw kwargs — go through with_config for consistency.
Trigger + schedule recipe. Cron schedule attached in UI: 0 * * * * hourly, 0 0 * * * daily. Event triggers: file arrival on S3 (sensor block), HTTP webhook (POST to /api/pipeline_schedules/N/pipeline_runs), API trigger (mage_ai.api.trigger_pipeline_run). Backfill via UI date range picker with max_concurrent_runs cap per pipeline. Sensor blocks poll external state (file exists, upstream pipeline done, Airbyte sync succeeded) and unblock downstream blocks on True return.
Fan-in join wiring rule. For any transformer with N upstream blocks, list them in upstream_blocks: in the same positional order as the function's parameters. def join(orders, customers, *args, **kwargs) requires upstream_blocks: [load_orders, load_customers] in that exact order. Swapping the YAML order without swapping the function signature silently produces a bogus join with no runtime error. Always double-check the mapping after every refactor.
Language-per-block choice. SQL blocks for GROUP BY / windowing / JOIN — the warehouse is faster, cheaper (no data egress), and the optimiser is smarter. Python (pandas / polars) blocks for row-level logic, custom libraries, and machine-learning transforms. PySpark blocks for big-data transforms that pandas can't hold. R blocks for the R-heavy data-science team. Mix languages within one pipeline — Mage passes DataFrames across language boundaries transparently.
Mage vs Dagster vs Prefect vs Airflow — matrix cold. Mage wins dev-loop UX + streaming; Dagster wins SDA lineage (asset graph); Prefect wins dynamic flows (.map + subflows); Airflow wins community + ecosystem scale (5000+ DAGs, largest connector library). Pick the tool whose winning axis matches the workload's load-bearing constraint. Multi-orchestrator posture (Mage + Airflow + Prefect) is a legitimate end-state, not a transition — use the right tool for each workload.
Custom connector recipe (30 lines). Any @data_loader that pages through an HTTP source: while loop over pagination cursor + requests.get with 429 (respect Retry-After header) + 5xx (exponential backoff) retry logic + return a DataFrame. Extract the HTTP + retry into a shared custom_sources/ module for reuse across pipelines. For sources you'll use in 3+ pipelines, promote to a proper Python package and publish to internal PyPI.
Airbyte + Mage coexistence post-2024. Two viable patterns: (a) sensor blocks that poll Airbyte connection status and unblock downstream Mage blocks on success; (b) native Mage loaders that call Airbyte connectors directly (roadmap converging 2025-2026). Retain Airbyte for the SaaS ingest catalogue you already run; layer Mage on top for downstream transforms and orchestration. The 2024 acquisition means both tools are one product family; investment in either compounds.
Common failure modes to pre-empt. Kafka source without enable_auto_commit=false → double-processing on failure; non-idempotent sink + at-least-once retries → data duplication; metadata.yaml upstream order mismatch → silent bogus join; io_config.yaml with raw secrets → git-history leak (always use references); DuckDB in-transformer for very large batches → OOM (bound batch size); scratchpad block accidentally scheduled → runs on cron. Every one has a two-line fix; catching them at authoring time saves 3 AM pages.
Migration cost estimates. Airflow → Mage: ~2 weeks per 10 DAGs (PythonOperator port, XCom removal). Prefect → Mage: ~1 week per 10 flows (paradigm parity high). dbt-only → Mage: ~1 week total (add loader/exporter around dbt). Dagster → Mage: not recommended (loses SDA lineage — the reason Dagster was picked). Budget the migration at 20-30% engineer-time for 3-6 months; expect payback in dev-loop savings within 6-9 months.

Frequently asked questions

What is Mage AI in one sentence?

Mage AI is a notebook-style, open-source data orchestrator where every pipeline is composed of typed blocks — @data_loader, @transformer, @data_exporter, @sensor, @custom — authored in a browser IDE with live output preview, wired into a DAG by implicit function-signature dependencies, versioned as first-class files in Git, executed by a single unified runtime for both batch and streaming, and (since Airbyte acquired Mage in 2024) roadmapped to converge with the ~350-connector Airbyte catalogue. It ships as open-source (Apache-2.0) with a managed offering called Mage Pro; the mental model interviewers probe for is "Airflow with the DAG-file-as-boilerplate replaced by decorator-per-block, plus a notebook IDE for interactive authoring, plus first-class streaming." Every senior data-engineering interview in 2026 probes Mage because the notebook-orchestrator category has proven itself and the four-tool comparison (Mage vs Dagster vs Prefect vs Airflow) is now a standard whiteboard question.

Mage vs Dagster — when do I pick each?

Pick Mage when your load-bearing constraint is iteration velocity (fast author-run-observe dev-loop), streaming (Kafka source → windowed transform → idempotent sink under ~100k events/sec), or integration coverage (90+ pre-built connectors plus the incoming Airbyte catalogue). Pick Dagster when your load-bearing constraint is software-defined-asset (SDA) lineage — regulated analytics engineering where "which asset produced customer 42's LTV?" is an audit-critical query. Dagster's asset graph is the primary abstraction: every asset knows its dependencies, its materialisation history, its data-quality checks. Mage's DAG is block-level, so "block A ran before block B" is queryable but not per-asset lineage. Concretely: a Series C analytics team ingesting Salesforce, running dbt, and adding real-time clickstream should pick Mage; a regulated fintech running compliance-audited daily attribution should pick Dagster. Both are valid answers; the wrong answer is picking either without probing the constraint first. Interviewers score highest on the candidate who says "depends on lineage requirement" before naming a tool.

What is the block DAG model?

The Mage block DAG model has three moving parts: (a) five block types — @data_loader (source; no upstream), @transformer (shape; ≥ 1 upstream), @data_exporter (sink; typically leaf), @sensor (blocks on external condition), @custom (scratchpad); (b) implicit dependencies — the downstream block's function signature takes one positional argument per upstream in the order declared in metadata.yaml, so if block C has two upstreams A and B its signature is def c(a_output, b_output, *args, **kwargs); (c) the wiring file — metadata.yaml lists every block and its upstream_blocks: [...] list, and Mage's runtime uses this graph to schedule blocks (independent blocks in parallel, dependent blocks in order). Compared to Airflow's DAG-file-plus-XCom paradigm, this is dramatically less code — no DAG(...) object construction, no python_callable=, no xcom_pull. Compared to Dagster's asset-graph paradigm, this is less semantically rich for lineage but faster to author. The one gotcha: getting the upstream_blocks: order out of sync with the function's parameter order silently produces the wrong data flow at runtime; always double-check after refactors.

Can Mage handle real streaming?

Yes — Mage ships a first-class streaming pipeline mode (type: streaming in metadata.yaml) with source blocks for Kafka, Kinesis, RabbitMQ, Amazon SQS, and Google Pub/Sub, transformer blocks that can embed DuckDB / Materialize / RisingWave for in-block SQL windowing, and sink blocks for Snowflake, BigQuery, S3, Kafka, and Postgres. The exactly-once story: source offsets are committed only after the sink acknowledges success, so any failure re-polls the same batch; combined with an idempotent sink (Snowflake MERGE, S3 with deterministic keys, Postgres ON CONFLICT, Kafka with keyed messages + log compaction), this yields effectively exactly-once semantics without a distributed transaction. The realistic ceiling is ~100k events/sec on a single Mage worker with < 2 GB of in-memory state; above that ceiling, or when complex event processing (CEP) is required, escalate to Apache Flink for its RocksDB state backend + cluster architecture. For workloads that fit under the ceiling — most clickstream sessionisation, IoT event enrichment, real-time dashboard aggregation — Mage streaming is dramatically simpler than standing up Flink and runs in the same operational plane as your batch pipelines.

How does Mage integrate with dbt?

Mage integrates dbt as a first-class block type — type: dbt in metadata.yaml, with pure-YAML configuration (dbt_project_path, command: build, select: +my_model, variables: {...}) and no Python wrapper needed. Under the hood it runs dbt build --select +my_model, but with Mage's scheduler wrapping the invocation: per-model status appears in the Mage UI (not just "block succeeded / failed"), retries are per-block, logging is captured, notifications fire on failure. This is qualitatively better than the Airflow equivalent (BashOperator calling dbt run in a shell) because Mage understands the dbt model list and can report per-model results. The typical pipeline shape: Mage @data_loader and @transformer blocks stage source data into the warehouse; a dbt block runs models tagged appropriately (e.g., select: tag:daily); Mage @data_exporter blocks read the post-dbt warehouse state and push reverse-ETL to Salesforce, HubSpot, Slack, etc. Prefer dbt build over separate dbt run + dbt test so model failures and test failures halt in one place. This pattern lets you keep your existing dbt project unchanged while gaining unified orchestration UI, cron scheduling, and per-model observability — a near-zero-risk migration from a BashOperator-plus-cron world.

What changed after Airbyte acquired Mage?

Airbyte acquired Mage in October 2024, and the roadmap since has been connector-catalogue convergence plus unified operational plane. Concretely: (a) Airbyte's ~350-connector catalogue is progressively landing as native Mage @data_loader blocks over 2025-2026, meaning workloads that today use an Airbyte connection and a Mage sensor block will migrate to a single Mage loader block; (b) both products remain independent open-source projects with their own commercial tiers (Airbyte Cloud, Mage Pro) but share a merged product roadmap; (c) teams evaluating Mage today are effectively also evaluating Airbyte's connector catalogue, which is the largest in the open-source ingest space. The strategic implication for a Series C data platform decision: betting on Mage in 2026 positions you for the merged product family, not against it — investment in either tool compounds. The coexistence pattern for the transition period is two-fold: (1) retain your Airbyte connections and add Mage sensor blocks that wait for Airbyte sync success before unblocking downstream Mage transforms; (2) as native Mage loaders for your specific connectors ship in the roadmap, migrate one connector at a time from Airbyte to Mage without changing the downstream DAG. This is a genuinely low-risk migration story — the acquisition has, so far, been additive rather than disruptive to either user base.

Practice on PipeCode

Drill the SQL practice library → for the warehouse-side transformation, join, and aggregation problems that Mage @transformer blocks and dbt models both spend most of their time on.
Rehearse system design against the design practice library → for the orchestrator-selection, migration, and multi-tool coexistence scenarios that separate architects from implementers in senior Mage interviews.
Sharpen the streaming axis with the streaming practice library → for the Kafka source, windowed aggregation, offset-commit, and idempotent-sink patterns that the Mage streaming block model is built around.
Practice the aggregation library → for the tumbling / sliding / session window problems that Mage streaming transformers most often need to implement in-block via DuckDB / Materialize / RisingWave.
Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-tool decision matrix against real graded inputs — dev-loop measurements, streaming latency budgets, lineage-requirement scenarios, and multi-orchestrator migration plans.

Lock in mage ai muscle memory

Docs explain the blocks. PipeCode drills explain the decision — when Mage's live-preview dev-loop earns its notebook-orchestrator premium, when Dagster's asset graph is the load-bearing lineage answer, when Prefect's dynamic mapping is the right per-tenant pattern, when Airflow's ecosystem still wins on legacy scale, when Mage streaming's ~100k events/sec ceiling forces a Flink escalation. Pipecode.ai is Leetcode for Data Engineering — orchestrator-first practice tuned for the production trade-offs senior data engineers actually defend in 2026 Mage AI interviews.

Practice SQL problems →
Practice streaming problems →

1. Why Mage matters in 2026

One notebook IDE, one block DAG, one runtime — the orchestrator that made ETL fun to write again

Worked example — the four-tool comparison table

Worked example — the "author → schedule" 10-minute dev-loop measurement

Worked example — the "pick the orchestrator" decision tree

Senior interview question on Mage adoption

Solution Using a workload-by-workload migration plan anchored on dev-loop and streaming wins

2. Block-based pipelines + DAG mental model

@data_loader → @transformer → @data_exporter — the block is the atom of every Mage DAG

Worked example — three-block Postgres → transform → S3 pipeline

Worked example — fan-in join of two loaders into one transformer

Worked example — SQL-language transformer against Snowflake

Senior interview question on block DAG design

Solution Using a fan-in loader graph, a SQL join transformer, and a dbt block for the downstream model

3. Streaming pipelines in Mage

Kafka source block → windowed transform → sink → commit offsets — the exactly-once streaming story most orchestrators get wrong

Worked example — Kafka clickstream sessionisation → Snowflake

Worked example — DuckDB tumbling-window aggregation

Worked example — when to pick Mage streaming vs a dedicated Flink cluster

Senior interview question on Mage streaming

Solution Using a Kafka source + module-state windowed transformer + Snowflake MERGE sink + post-sink offset commit

4. Integrations + destinations

90+ pre-built connectors, first-class dbt, custom Python sources — the integration layer that folds Airbyte's catalogue into the block DAG

Worked example — Salesforce → Snowflake with a dbt transformation stage

Worked example — custom Python source for a proprietary internal API

Worked example — the io_config.yaml secrets pattern

Senior interview question on Mage integrations

Solution Using dbt block for models + Airbyte-connector-backed loaders + a phased migration anchored on the acquisition roadmap

5. Mage vs Dagster / Prefect / Airflow + interview signals

The four-tool decision matrix — Mage wins hands-on UX, Dagster wins SDA lineage, Prefect wins dynamic flows, Airflow wins legacy scale

Worked example — the four-axis comparison whiteboard

Worked example — the "when Mage is the wrong pick" scenarios

Worked example — the five interview questions to rehearse

Senior interview question on Mage vs the field

Solution Using a workload-scored migration plan anchored on Mage for greenfield and Airflow-stay for legacy

Cheat sheet — Mage recipes

Frequently asked questions

What is Mage AI in one sentence?

Mage vs Dagster — when do I pick each?

What is the block DAG model?

Can Mage handle real streaming?

How does Mage integrate with dbt?

What changed after Airbyte acquired Mage?

Practice on PipeCode

Lock in mage ai muscle memory

`@data_loader` → `@transformer` → `@data_exporter` — the block is the atom of every Mage DAG

Worked example — the `io_config.yaml` secrets pattern