You'll start by meeting the Airflow components, writing your first Dags with the TaskFlow API, and passing data between tasks with XCom.

Airflow core concepts

Writing a minimal TaskFlow Dag

Matching Airflow components to their roles

The TaskFlow API

Converting a classic Dag to TaskFlow

Adding a task and wiring dependencies

Exploring a pipeline in the Airflow UI

Sharing data between tasks

Returning data from a TaskFlow function

Connecting a classic operator to a TaskFlow task

Understanding XCom limitations

Authoring Dags with TaskFlow and XCom

From there, you'll run tasks in parallel with dynamic task mapping, schedule Dags by data with Assets, and add human approval steps.

Dynamic Task Mapping

Processing multiple endpoints with expand

Fixing a shared parameter with partial

Pairing inputs with zip

Data-aware scheduling with Assets

Signaling data availability with an Asset outlet

Scheduling a consumer Dag on an Asset

Seeing asset-based scheduling in action

Templating, idempotency, and backfilling

Adding date-aware templating

Making a pipeline idempotent

Triggering a backfill

Human-in-the-loop workflows

Collecting form input with HITLEntryOperator

Responding to a Required Action in the UI

Branching with HITLBranchOperator

Dynamic and Data-Aware Pipelines

In this chapter, you'll handle failures with retries and callbacks, save resources with deferrable sensors, and test your Dags at three levels.

Robust Dags with failure handling and retries

Configuring retries with exponential backoff

Auto-pausing a failing Dag

Choosing the right failure handling strategy

Deferrable operators

Configuring a sensor for production

Making a sensor deferrable

When to use poke, reschedule, or deferrable

Testing Airflow code

Writing a Dag integrity test

Unit testing a task function

Integration testing with dag.test()

Organizing complex Dags with Task Groups

Grouping related tasks

Nesting task groups with display names

Organizing tasks by domain

Preparing Dags for Production

In this final chapter, you'll build a SQL ETL pipeline on DuckDB, add partition-aware scheduling with Asset Partitions, and embed data quality checks.

Running SQL workloads

from airflow.sdk import dag, task
# Import SQLExecuteQueryOperator
from airflow.providers.common.sql.operators.sql import ____


@dag(schedule="@daily")
def sales_pipeline():

    # Aggregate raw orders into the daily summary table
    aggregate = ____(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="""
            INSERT INTO daily_summary (order_date, total_orders, total_revenue)
            SELECT order_date, COUNT(*), SUM(revenue)
            FROM raw_orders
            GROUP BY order_date
        """,
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="____",
        conn_id="____",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline()

if __name__ == "__main__":
    pipeline.test()


from airflow.sdk import dag, task
# Import SQLExecuteQueryOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


@dag(schedule="@daily")
def sales_pipeline():

    # Aggregate raw orders into the daily summary table
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="""
            INSERT INTO daily_summary (order_date, total_orders, total_revenue)
            SELECT order_date, COUNT(*), SUM(revenue)
            FROM raw_orders
            GROUP BY order_date
        """,
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline()

if __name__ == "__main__":
    pipeline.test()


Aggregating sales data with SQL

DELETE FROM daily_summary WHERE order_date = '____';
INSERT INTO daily_summary (order_date, total_orders, total_revenue)
SELECT order_date, COUNT(*), SUM(revenue)
FROM raw_orders
WHERE order_date = '____'
GROUP BY order_date;


from airflow.sdk import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


# Set the template search path, so Airflow finds SQL files
@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/____")
def sales_pipeline_v2():

    # Reference the external SQL file instead of inline SQL
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="____.sql",
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v2()

if __name__ == "__main__":
    pipeline.test()


DELETE FROM daily_summary WHERE order_date = '{{ ds }}';
INSERT INTO daily_summary (order_date, total_orders, total_revenue)
SELECT order_date, COUNT(*), SUM(revenue)
FROM raw_orders
WHERE order_date = '{{ ds }}'
GROUP BY order_date;


from airflow.sdk import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


# Set the template search path, so Airflow finds SQL files
@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v2():

    # Reference the external SQL file instead of inline SQL
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v2()

if __name__ == "__main__":
    pipeline.test()


Parameterizing SQL and extracting it into files

Choosing the safe approach for user input

Partition-aware scheduling with Asset Partitions

# Import CronPartitionTimetable and Asset
from airflow.sdk import dag, task, ____, ____
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# Define the daily_sales asset
daily_sales = ____("daily_sales")


# Schedule with CronPartitionTimetable
@dag(schedule=____("0 0 * * *", ____="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        # Emit a partitioned asset event
        outlets=[____],
        sql="aggregate_sales.sql",
    )

    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    load >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v3()

if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        pipeline.test()


# Import CronPartitionTimetable and Asset
from airflow.sdk import dag, task, CronPartitionTimetable, Asset
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# Define the daily_sales asset
daily_sales = Asset("daily_sales")


# Schedule with CronPartitionTimetable
@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        # Emit a partitioned asset event
        outlets=[daily_sales],
        sql="aggregate_sales.sql",
    )

    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    load >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v3()

if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        pipeline.test()


Creating a partitioned pipeline

# Import PartitionedAssetTimetable and StartOfDayMapper
from airflow.sdk import dag, task, Asset, CronPartitionTimetable, ____, ____
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

daily_sales = Asset("daily_sales")


# --- Upstream: partitioned sales load (from previous exercise) ---
@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():
    SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        outlets=[daily_sales],
        sql="aggregate_sales.sql",
    )

upstream = sales_pipeline_v3()


# --- Downstream: report triggered by partitioned asset event ---
# Schedule on partitioned daily_sales events with StartOfDayMapper
@dag(schedule=____(
    assets=____,
    partition_mapper_config={daily_sales: ____()}
))
def sales_report():

    @task
    def generate_report(dag_run=None):
        # Read the partition key from the Dag run
        partition_key = dag_run.____
        print(f"Generating report for partition: {partition_key}")

    generate_report()

downstream = sales_report()


if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        upstream.test()


# Import PartitionedAssetTimetable and StartOfDayMapper
from airflow.sdk import dag, task, Asset, CronPartitionTimetable, PartitionedAssetTimetable, StartOfDayMapper
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

daily_sales = Asset("daily_sales")


# --- Upstream: partitioned sales load (from previous exercise) ---
@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():
    SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        outlets=[daily_sales],
        sql="aggregate_sales.sql",
    )

upstream = sales_pipeline_v3()


# --- Downstream: report triggered by partitioned asset event ---
# Schedule on partitioned daily_sales events with StartOfDayMapper
@dag(schedule=PartitionedAssetTimetable(
    assets=daily_sales,
    partition_mapper_config={daily_sales: StartOfDayMapper()}
))
def sales_report():

    @task
    def generate_report(dag_run=None):
        # Read the partition key from the Dag run
        partition_key = dag_run.partition_key
        print(f"Generating report for partition: {partition_key}")

    generate_report()

downstream = sales_report()


if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        upstream.test()


Scheduling on partitioned assets

Watching partitioned assets trigger a downstream Dag

Data quality checks

from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    # Import SQLColumnCheckOperator
    ____,
)


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v4():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    # Validate column-level data quality after loading
    check_columns = ____(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"____": {"greater_than": 0}},
            "total_orders": {"____": {"equal_to": 0}},
        },
    )

    load >> check_columns


pipeline = sales_pipeline_v4()

if __name__ == "__main__":
    pipeline.test()


from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    # Import SQLColumnCheckOperator
    SQLColumnCheckOperator,
)


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v4():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    # Validate column-level data quality after loading
    check_columns = SQLColumnCheckOperator(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"min": {"greater_than": 0}},
            "total_orders": {"null_check": {"equal_to": 0}},
        },
    )

    load >> check_columns


pipeline = sales_pipeline_v4()

if __name__ == "__main__":
    pipeline.test()


Adding column-level quality checks

from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    SQLColumnCheckOperator,
    SQLTableCheckOperator,
)


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v5():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    check_columns = SQLColumnCheckOperator(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"min": {"greater_than": 0}},
            "total_orders": {"null_check": {"equal_to": 0}},
        },
    )

    # Add table-level quality check
    check_table = ____(
        task_id="check_table",
        conn_id="duckdb_analytics",
        table="daily_summary",
        # Verify that the table has at least one row
        checks={
            "row_count_check": {"check_statement": "COUNT(*) ____ ____"},
        },
    )

    load >> check_columns >> check_table


pipeline = sales_pipeline_v5()

if __name__ == "__main__":
    pipeline.test()


from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    SQLColumnCheckOperator,
    SQLTableCheckOperator,
)


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v5():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    check_columns = SQLColumnCheckOperator(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"min": {"greater_than": 0}},
            "total_orders": {"null_check": {"equal_to": 0}},
        },
    )

    # Add table-level quality check
    check_table = SQLTableCheckOperator(
        task_id="check_table",
        conn_id="duckdb_analytics",
        table="daily_summary",
        # Verify that the table has at least one row
        checks={
            "row_count_check": {"check_statement": "COUNT(*) > 0"},
        },
    )

    load >> check_columns >> check_table


pipeline = sales_pipeline_v5()

if __name__ == "__main__":
    pipeline.test()


Adding table-level quality checks

Catching data issues with the right operator

From development to production

Triggering a pipeline from the CLI

Backfilling historical data

Options

Build

Observe

Production challenges with Airflow

Congratulations!

Building a Production SQL ETL Pipeline

Want to take your Airflow skills further? This course uses Airflow 3.2, so you'll learn the latest way of doing things. You'll author Dags with the TaskFlow API, schedule them based on data using Assets and the new Asset Partitions, and make them reliable with retries, callbacks, and tests.

In the final chapter, you'll build an end-to-end SQL ETL pipeline on DuckDB and add data quality checks directly in Airflow, without any third-party libraries, so the data your pipeline produces stays trustworthy. By the end, you'll know how to take a pipeline from a prototype to something you can actually run in production.

Introduction to Apache Airflow in Python

Author Dags with the TaskFlow API, asset-based scheduling, and deferrable sensors, and run an end-to-end SQL ETL pipeline with quality checks.



Building Data Pipelines with Airflow

Senior Developer Advocate at Astronomer

When ds is not enough

CronPartitionTimetable

Accessing partition keys

Partitioned asset events

PartitionedAssetTimetable

StartOfDayMapper

The full flow

Let's practice!

https://s3.amazonaws.com/assets.datacamp.com/production/course_49741/subtitles/course_49741_a88a2af4f4bcde0b7fb7f759c06f59ce.vtt

Building a Production SQL ETL Pipeline - Partition-aware scheduling with Asset Partitions

Partition-aware scheduling with Asset Partitions

Create Your Free Account