Building a Pipeline DAG

A full worked DAG for an orders pipeline — watermark, interval-aware load, parallel marts, a rollup, a quality check, and a failure alert — with the complete file at the end.

Intermediate

The earlier articles covered each Airflow concept in isolation. This one builds one realistic DAG that uses all of them together, explaining each decision, with the complete file at the end.

The pipeline: every hour, load new order events into staging.orders, rebuild three marts from staging, roll the marts up over the last 30 days, check the result, and alert if anything failed.

The shape#

Text
read_watermark      |load_raw_to_staging      |      +--> build_customer_metrics --+      +--> build_daily_revenue -----+--> rollup_last_30_days --> check_rollup      +--> build_segment_revenue ---+
(on any failure)  alert_on_failure
  • read_watermark finds where the last run stopped and passes it forward (XCom).
  • load_raw_to_staging loads only the current interval's rows, idempotently.
  • The three mart tasks are independent, so they run in parallel.
  • rollup_last_30_days waits for all three.
  • check_rollup is a data-quality gate.
  • alert_on_failure is a DAG-level failure callback, not a task.

The callables live outside the DAG#

Every unit of real work is a plain function in its own module — no Airflow imports, independently testable. The DAG file only wires them together.

Python
# common/clickhouse.pyfrom airflow.hooks.base import BaseHookfrom clickhouse_connect import get_client

def clickhouse_client():    conn = BaseHook.get_connection("clickhouse_default")    return get_client(        host=conn.host, port=conn.port,        username=conn.login, password=conn.password,    )

Credentials come from the clickhouse_default Connection, not from the DAG.

Task 1: read the watermark#

Find the newest created_at already in staging. Returning the value pushes it to XCom:

Python
# transform/watermark.pyfrom common.clickhouse import clickhouse_client

def read_staging_watermark(**context):    client = clickhouse_client()    result = client.query("SELECT maxOrNull(created_at) FROM staging.orders")    watermark = result.result_columns[0][0]    print(f"staging watermark: {watermark}")    return watermark        # -> XCom "return_value"

A watermark is a single timestamp — small, so XCom is the right channel. A full dataset would not be.

Task 2: interval-aware, idempotent load#

Load rows for this run's interval, and make the write safe to repeat:

Python
# transform/raw_to_staging.pyfrom common.clickhouse import clickhouse_client

def load_raw_to_staging(**context):    start = context["data_interval_start"]    end = context["data_interval_end"]    client = clickhouse_client()
    # idempotent: clear this window first, then insert it    client.command(        "ALTER TABLE staging.orders DELETE WHERE created_at >= {s:DateTime} AND created_at < {e:DateTime}",        parameters={"s": start, "e": end},    )    client.command(        """        INSERT INTO staging.orders        SELECT order_id, customer_id, created_at, amount, status        FROM raw.orders        WHERE created_at >= {s:DateTime} AND created_at < {e:DateTime}        """,        parameters={"s": start, "e": end},    )

Why this shape:

  • data_interval_start / data_interval_end, not datetime.now(), so a late run or a backfill processes the correct window (Scheduling and Backfill).
  • Delete-then-insert the window, so a retry or a backfill overwrites instead of duplicating (Building Staging and Data Marts).
  • The watermark from task 1 is available via context["ti"].xcom_pull(task_ids="read_watermark") if the load wants a belt-and-braces lower bound; the interval is the primary boundary.

Task 3: three marts in parallel#

Each mart is its own function and its own task. They do not depend on each other, so Airflow runs them concurrently:

Python
# marts/build.pyfrom common.clickhouse import clickhouse_client

def _rebuild_day(table, select_sql, day):    client = clickhouse_client()    # table names are identifiers (built into the string); day is a value (bound)    client.command(        f"ALTER TABLE {table} DROP PARTITION " + "{day:String}",        parameters={"day": day},    )    client.command(f"INSERT INTO {table} {select_sql}")

def build_customer_metrics(**context):    day = context["ds"]    _rebuild_day("mart.customer_metrics", "SELECT ... FROM staging.orders WHERE ...", day)

def build_daily_revenue(**context):    day = context["ds"]    _rebuild_day("mart.daily_revenue", "SELECT ... FROM staging.orders WHERE ...", day)

def build_segment_revenue(**context):    day = context["ds"]    _rebuild_day(        "mart.segment_revenue",        "SELECT ... FROM staging.orders o LEFT JOIN staging.customers c ON ...",        day,    )

Same idempotency pattern: drop the day's partition, then insert it.

Task 4: the rollup#

Runs once, after all three marts, over a longer window:

Python
# marts/rollup.pyfrom common.clickhouse import clickhouse_client

def rollup_last_30_days(**context):    end = context["data_interval_end"]    client = clickhouse_client()    client.command(        """        INSERT INTO mart.customer_30d        SELECT customer_id, sum(order_count), sum(total_amount)        FROM mart.customer_metrics        WHERE day >= {e:DateTime} - INTERVAL 30 DAY        GROUP BY customer_id        """,        parameters={"e": end},    )

Task 5: the quality gate#

A task that raises if the output is wrong, so the run fails loudly instead of publishing bad data:

Python
# checks/rollup.pyfrom common.clickhouse import clickhouse_client

def check_rollup(**context):    client = clickhouse_client()    rows = client.query("SELECT count() FROM mart.customer_30d").result_columns[0][0]    if rows == 0:        raise ValueError("mart.customer_30d is empty after the rollup")

The DAG file#

Everything above is plain functions. The DAG file imports them and declares the graph:

Python
# dags/orders_pipeline.pyfrom datetime import datetime, timedelta
from airflow import DAGfrom airflow.operators.python import PythonOperator
from transform.watermark import read_staging_watermarkfrom transform.raw_to_staging import load_raw_to_stagingfrom marts.build import build_customer_metrics, build_daily_revenue, build_segment_revenuefrom marts.rollup import rollup_last_30_daysfrom checks.rollup import check_rollupfrom common.alerts import alert_on_failure

default_args = {    "owner": "data-team",    "retries": 2,    "retry_delay": timedelta(minutes=5),    "on_failure_callback": alert_on_failure,}
with DAG(    dag_id="orders_pipeline",    default_args=default_args,    description="Hourly: raw -> staging -> marts -> 30d rollup",    start_date=datetime(2026, 1, 1),    schedule="@hourly",    catchup=False,    max_active_runs=1,    tags=["orders", "etl"],) as dag:
    read_watermark = PythonOperator(        task_id="read_watermark",        python_callable=read_staging_watermark,    )
    load_staging = PythonOperator(        task_id="load_raw_to_staging",        python_callable=load_raw_to_staging,    )
    marts = [        PythonOperator(task_id="build_customer_metrics", python_callable=build_customer_metrics),        PythonOperator(task_id="build_daily_revenue", python_callable=build_daily_revenue),        PythonOperator(task_id="build_segment_revenue", python_callable=build_segment_revenue),    ]
    rollup = PythonOperator(        task_id="rollup_last_30_days",        python_callable=rollup_last_30_days,    )
    check = PythonOperator(        task_id="check_rollup",        python_callable=check_rollup,    )
    read_watermark >> load_staging >> marts >> rollup >> check

The last line is the whole structure: a chain, a fan-out to the marts list, a fan-in to rollup, then check. on_failure_callback in default_args fires alert_on_failure for any task that fails all its retries.

What each concept contributed#

ConceptWhere it shows up
DAG + default_argsdag_id, retries, on_failure_callback
schedule + catchup=False@hourly, no backfill storm on enable
Interval-aware tasksdata_interval_start/end, ds in the callables
Idempotent tasksdelete-window / drop-partition before insert
PythonOperator + external functionsevery task points at a plain, testable function
Dependencies>> marts >> rollup >> check fan-out / fan-in
XComread_watermark returns a value the load can pull
Connections / Hooksclickhouse_client() reads clickhouse_default
Operabilityon_failure_callback, quality gate, max_active_runs=1

Testing it#

Bash
airflow dags test orders_pipeline 2026-03-01T00:00:00airflow tasks test orders_pipeline load_raw_to_staging 2026-03-01T00:00:00

airflow tasks test runs one callable against a real interval with no metadata writes — the fastest loop for developing the functions.

Common mistakes#

Logic creeping into the DAG file#

The moment a transformation lives in dags/orders_pipeline.py, it runs on every parse and cannot be unit-tested. Keep the file to imports and >>.

A mart task that appends#

Without the DROP PARTITION / DELETE WHERE, a retry doubles the day. Every task in a scheduled DAG must be safe to run twice.

rollup before the marts finish#

load_staging >> marts >> rollup makes rollup wait for all three. Chaining it after only one mart lets it run on partial data.

No failure alert#

on_failure_callback (or email_on_failure) is the difference between noticing a broken run and a stakeholder noticing an empty dashboard.

Backfilling without max_active_runs=1#

Parallel backfill runs writing the same partitions collide. Serialise them.

See also#

Building Staging and Data Marts
The SQL inside the callables — INSERT ... SELECT, JOIN, idempotent loads.
Running and Operating Airflow
Watching this DAG run and recovering when a task fails.