Chapter 11 · Operating It

Orchestration with Airflow

The DAG is the table of contents of the pipeline. It contains no SQL and no data logic, only which file runs, on which connection, after what. This chapter walks through it step by step, then opens up the ~100-line plugin that does the actual talking to SQL Server.

The whole DAG

"""
### SBEcoLTDDW end-to-end pipeline

Loads the **SBEcoLTDDW** data warehouse (SQL Server 2022 for Linux, in a Docker container) from
its source system, AdventureWorks2022 (sales OLTP):

| Source table(s) | Warehouse table | How |
|---|---|---|
| Sales.Customer + Person.* | `Dimension.Customer` | full extract, SCD Type 1 + 2 |
| Sales.SalesOrderHeader | `Fact.Orders` | incremental watermark, idempotent MERGE |
| (generated) | `Dimension.Date` | calendar 2010-2030 |

**This file only orchestrates.** Every task runs one `.sql` file from `/opt/airflow/sql/sbecoltddw`
(mounted from `~/airflow/sql` on the Linux host). Open a task and look at *Rendered Template* to see
the exact SQL.

Steps: `deploy` → `start_run` → `extract` → `stage_checks` → `dimensions` → `facts` → `post_load`
→ `dq_gate` → `publish` → `end_run` (or `fail_run` when anything failed).

Trigger with `{"full_reload": true}` to ignore the watermark and reload all orders.
"""
from pathlib import Path

import pendulum
from airflow import DAG
from airflow.models.param import Param
from airflow.utils.task_group import TaskGroup
from airflow.utils.trigger_rule import TriggerRule

from sbecoltddw_operators import MsSqlFileOperator

SQL_ROOT = "/opt/airflow/sql/sbecoltddw"
ETL = "mssql_sbecoltddw_etl"      # login DataLoader -> Resource Governor group DataLoads


def sql_task(task_id, sql, conn_id=ETL, **kwargs):
    """One task = one SQL file."""
    return MsSqlFileOperator(task_id=task_id, sql=sql, conn_id=conn_id, **kwargs)


with DAG(
    dag_id="sbecoltddw_e2e_pipeline",
    description="SBEcoLTDDW: source -> staging -> dimensions -> facts -> DQ gate -> reporting",
    start_date=pendulum.datetime(2026, 10, 1, tz="UTC"),
    schedule=None,
    catchup=False,
    max_active_runs=1,
    template_searchpath=[SQL_ROOT],
    params={"full_reload": Param(False, type="boolean", description="Ignore the order watermark and reload everything")},
    default_args={"owner": "sbecoltddw", "retries": 0},
    doc_md=__doc__,
    tags=["sbecoltddw", "sqlserver", "data-warehouse"],
) as dag:

    # 1. deploy: create / upgrade every pipeline object - one task per file, in file order
    with TaskGroup("deploy", tooltip="Idempotent DDL: schemas, control tables, staging, DW changes, views") as deploy:
        files = sorted(Path(SQL_ROOT, "01_deploy").glob("*.sql"))
        previous = None
        for f in files:
            t = sql_task(f.stem, f"01_deploy/{f.name}")
            if previous:
                previous >> t
            previous = t

    # 2. register the run in audit.BatchRun
    start_run = sql_task("start_run", "00_control/start_run.sql")

    # 3. extract: source -> stg (independent, so they run in parallel)
    with TaskGroup("extract", tooltip="Copy source data into the stg schema") as extract:
        extract_customers = sql_task("extract_customers", "02_extract/extract_customers.sql")
        extract_sales_orders = sql_task("extract_sales_orders", "02_extract/extract_sales_orders.sql")

    # 4. stop early if staging is unusable
    stage_checks = sql_task("stage_checks", "03_stage_checks/check_staging.sql")

    # 5. dimensions before facts (facts look up dimension keys)
    with TaskGroup("dimensions", tooltip="Date (generated), Customer (SCD Type 1 + 2)") as dimensions:
        load_dim_date = sql_task("load_dim_date", "04_dimensions/load_dim_date.sql")
        load_dim_customer = sql_task("load_dim_customer_scd2", "04_dimensions/load_dim_customer_scd2.sql")

    # 6. facts
    with TaskGroup("facts", tooltip="Fact.Orders (incremental, idempotent MERGE)") as facts:
        load_fact_orders = sql_task("load_fact_orders", "05_facts/load_fact_orders.sql")

    # 7. post-load maintenance
    with TaskGroup("post_load", tooltip="Columnstore, statistics, indexed view") as post_load:
        maintain_columnstore = sql_task("maintain_columnstore", "06_post_load/maintain_columnstore.sql")
        update_statistics = sql_task("update_statistics", "06_post_load/update_statistics.sql")
        check_indexed_view = sql_task("check_indexed_view", "06_post_load/check_indexed_view.sql")
        maintain_columnstore >> update_statistics >> check_indexed_view

    # 8. the data quality gate
    dq_gate = sql_task("dq_gate", "07_dq/dq_gate.sql")

    # 9. publish: commit the watermark only after everything passed
    with TaskGroup("publish", tooltip="Advance watermarks, summarise the run") as publish:
        advance_watermarks = sql_task("advance_watermarks", "08_publish/advance_watermarks.sql")
        run_summary = sql_task("run_summary", "08_publish/run_summary.sql")
        advance_watermarks >> run_summary

    # 10. close the batch
    end_run = sql_task("end_run", "00_control/end_run.sql")
    fail_run = sql_task("fail_run", "00_control/fail_run.sql", trigger_rule=TriggerRule.ONE_FAILED)

    deploy >> start_run >> extract >> stage_checks >> dimensions >> facts >> post_load >> dq_gate >> publish
    publish >> [end_run, fail_run]

Step by step

1. deploy: a task per file, discovered at parse time. The group globs 01_deploy/*.sql and chains one task per file in name order. When 11_ivw_sales_by_year.sql was added (Chapter 8), the next run simply had one more task. No Python changed. Every file is idempotent, so deploying on every run is cheap (0.2–0.4 s each) and keeps the schema in step with the code. 01_deploy/01–11 · Chapters 3, 5, 6, 8, 13
2. start_run: register the batch. Inserts (or, on a cleared re-run, resets) the audit.BatchRun row keyed by Airflow's run_id. Every later file finds its batch through {{ run_id }}. 00_control/start_run.sql
3. extract: two independent tasks. Customers (full) and orders (incremental) are copied from AdventureWorks2022 in parallel; both are cross-database INSERT … SELECTs inside the SQL Server Linux container. 02_extract · Chapter 5
4. stage_checks: fail fast. One task, downstream of the whole extract group, so it sees both staged tables. 03_stage_checks · Chapter 5
5. dimensions, then facts. The group-to-group arrow is the important dependency: no fact can load before every dimension is ready. Inside each group the tasks are independent. 04_dimensions, 05_facts · Chapters 6, 7
6. post_load: a chain. Compress columnstore, then update statistics (so they describe the compressed table), then prove the indexed view matches. 06_post_load · Chapter 8
7. dq_gate: the decision point. If it fails, everything after it becomes upstream_failed. 07_dq · Chapter 10
8. publish: commit. The watermark advances, and the run summary is logged. 08_publish · Chapter 5
9. end_run or fail_run. end_run uses the default all_success rule. fail_run uses one_failed: it runs only if its upstream (run_summary) failed or became upstream_failed, and marks the batch Failed. Exactly one of the two runs; the other is skipped or upstream_failed, so the DAG run's final state is still correct. 00_control/end_run.sql, fail_run.sql

Jinja in the SQL files

Because the operator declares template_ext = (".sql",), Airflow treats any sql argument ending in .sql as a file to load from template_searchpath and render with Jinja. The SQL files use only a handful of template values:

Open any task instance in the Airflow UI and choose Rendered Template to see the exact T-SQL that ran, ready to paste into SSMS.

The operator: ~100 lines of plumbing

"""
Reusable Airflow building blocks for the SBEcoLTDDW pipeline.

Orchestration plumbing only - no business SQL lives here. Every task points at a .sql file
under /opt/airflow/sql (mounted from ~/airflow/sql), so the SQL can be read, reviewed and
run on its own.

MsSqlFileOperator   runs one T-SQL batch (a .sql file, rendered by Jinja) on SQL Server,
                    writes every PRINT message and result set to the task log, and fails
                    the task if any statement in the batch raises an error.
"""
import datetime as dt
import decimal

import pyodbc
from airflow.exceptions import AirflowException
from airflow.hooks.base import BaseHook
from airflow.models import BaseOperator


def mssql_connect(conn_id):
    """Open a pyodbc connection from an Airflow connection (host, port, schema = database, login)."""
    c = BaseHook.get_connection(conn_id)
    pwd = "{" + (c.password or "").replace("}", "}}") + "}"
    return pyodbc.connect(
        "DRIVER={ODBC Driver 18 for SQL Server};"
        f"SERVER={c.host},{c.port or 1433};DATABASE={c.schema};UID={c.login};PWD={pwd};"
        "Encrypt=yes;TrustServerCertificate=yes;",
        autocommit=True,
    )


def _plain(value):
    """Make a result value JSON-friendly for XCom."""
    if isinstance(value, (dt.date, dt.datetime, dt.time, decimal.Decimal)):
        return str(value)
    if isinstance(value, (bytes, bytearray)):
        return "0x" + value.hex()
    return value


def _format_table(columns, rows, limit=50):
    shown = [[("NULL" if v is None else str(_plain(v))) for v in r] for r in rows[:limit]]
    widths = [max([len(c)] + [len(r[i]) for r in shown]) for i, c in enumerate(columns)]
    line = "+".join("-" * (w + 2) for w in widths)
    out = [" | ".join(c.ljust(w) for c, w in zip(columns, widths)), line]
    out += [" | ".join(v.ljust(w) for v, w in zip(r, widths)) for r in shown]
    if len(rows) > limit:
        out.append(f"... {len(rows) - limit} more rows")
    return "\n".join(out)


def run_batch(conn_id, sql, log):
    """Execute one batch and walk every result; returns the last result set as a list of dicts."""
    conn = mssql_connect(conn_id)
    cur = conn.cursor()
    last = []
    try:
        cur.execute(sql)
        n = 0
        while True:
            for _, msg in getattr(cur, "messages", None) or []:
                log.info("SQL> %s", msg.split("[SQL Server]")[-1])
            if cur.description:
                n += 1
                cols = [d[0] for d in cur.description]
                rows = cur.fetchall()
                log.info("result set %d (%d rows):\n%s", n, len(rows), _format_table(cols, rows))
                last = [{c: _plain(v) for c, v in zip(cols, r)} for r in rows[:50]]
            # nextset() raises if a later statement in the batch failed (e.g. THROW in a check)
            if not cur.nextset():
                break
    except pyodbc.Error as exc:
        raise AirflowException(f"SQL batch failed: {exc}") from exc
    finally:
        conn.close()
    return last


class MsSqlFileOperator(BaseOperator):
    """Run a .sql file (one T-SQL batch, no GO) on SQL Server."""

    template_fields = ("sql",)
    template_ext = (".sql",)
    template_fields_renderers = {"sql": "sql"}
    ui_color = "#dbeafe"

    def __init__(self, *, sql, conn_id, **kwargs):
        super().__init__(**kwargs)
        self.sql = sql
        self.conn_id = conn_id

    def execute(self, context):
        self.log.info("connection: %s", self.conn_id)
        return run_batch(self.conn_id, self.sql, self.log)

Three behaviours make it suitable for warehouse SQL:

• It walks every result set. A T-SQL batch returns one result per SELECT, and an error raised by a later statement (like the gate's THROW) only surfaces when the client asks for the next result. The while … cur.nextset() loop guarantees every error fails the task.
• It logs PRINT messages and tables. cursor.messages carries PRINT output, so the task log reads like an SSMS Messages tab. That is where every output in this tutorial was captured from.
• It returns the last result set (up to 50 rows, JSON-safe) as the task's XCom, handy for a downstream branch or an alert.

The simulator DAG: branching

sbecoltddw_source_changes plays the source application. A BranchPythonOperator picks one path from the scenario param, the other paths are skipped, and show_after uses none_failed_min_one_success so it still runs after a skipped branch.

    show_before = MsSqlFileOperator(task_id="show_before", sql="90_source_changes/show_customer.sql", conn_id=SOURCE)
    scenario = BranchPythonOperator(task_id="choose_scenario", python_callable=choose_scenario)

    move_customer = MsSqlFileOperator(task_id="move_customer", sql="90_source_changes/move_customer.sql", conn_id=SOURCE)
    insert_new_orders = MsSqlFileOperator(task_id="insert_new_orders", sql="90_source_changes/insert_new_orders.sql", conn_id=SOURCE)
    insert_bad_order = MsSqlFileOperator(task_id="insert_bad_order", sql="90_source_changes/insert_bad_order.sql", conn_id=SOURCE)
    fix_bad_order = MsSqlFileOperator(task_id="fix_bad_order", sql="90_source_changes/fix_bad_order.sql", conn_id=SOURCE)

    show_after = MsSqlFileOperator(
        task_id="show_after", sql="90_source_changes/show_customer.sql", conn_id=SOURCE,
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
    )

    show_before >> scenario >> [move_customer, insert_bad_order, fix_bad_order]
    move_customer >> insert_new_orders
    [insert_new_orders, insert_bad_order, fix_bad_order] >> show_after