首先，您将认识 Airflow 的各组成部分，使用 TaskFlow API 编写首个 Dags，并通过 XCom 在任务间传递数据。

Airflow 核心概念

编写最小化的 TaskFlow Dag

将 Airflow 组件与其角色匹配

TaskFlow API

将经典 Dag 转换为 TaskFlow

添加任务并连接依赖关系

在 Airflow UI 中探索一个管道

在任务之间共享数据

从 TaskFlow 函数返回数据

将经典算子连接到 TaskFlow 任务

理解 XCom 的限制

使用 TaskFlow 与 XCom 编写 Dags

接着，您将通过动态任务映射并行运行任务，使用 Assets 按数据调度 Dags，并加入人工审批步骤。

动态任务映射

使用 expand 处理多个端点

用 partial 固定共享参数

使用 zip 配对输入

基于 Assets 的数据感知调度

使用 Asset 输出标记数据可用性

在 Asset 上调度消费者 Dag

直观体验基于 Asset 的调度

模板化、幂等性与回填

添加日期感知的模板

让管道具备幂等性

触发回填

Human-in-the-loop 工作流

使用 HITLEntryOperator 收集表单输入

在 UI 中响应 Required Action

使用 HITLBranchOperator 实现分支

动态且数据感知的流水线

本章中，您将通过重试与回调处理失败，使用可延迟传感器节省资源，并在三个层级测试您的 Dags。

通过失败处理与重试打造健壮的 Dags

使用指数退避配置重试

自动暂停失败的 Dag

选择合适的故障处理策略

可延迟运算符

为生产环境配置传感器

将传感器改为可延迟执行（deferrable）

何时使用 poke、reschedule 或 deferrable

测试 Airflow 代码

编写 Dag 完整性测试

对任务函数进行单元测试

使用 dag.test() 进行集成测试

使用 Task Groups 组织复杂的 Dags

将相关任务分组

使用可见名称嵌套任务组

按领域组织任务

为生产环境准备 Dags

在最后一章，您将基于 DuckDB 构建 SQL ETL 流水线，使用 Asset Partitions 实现分区感知调度，并内嵌数据质量检查。

运行 SQL 工作负载

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()


使用 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()


为 SQL 参数化并将其拆分到文件中

为用户输入选择安全的方法

使用 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()


创建分区管道

# 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()


在分区资产上进行调度

观察分区资产如何触发下游 Dag

数据质量检查

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()


添加列级质量检查

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()


添加表级数据质量检查

用合适的运算符捕捉数据问题

从开发到生产

从 CLI 触发一条流水线

回填历史数据

Options

Build

Observe

Airflow 的生产环境挑战

恭喜您！

构建生产级 SQL ETL 流水线

想进一步提升您的 Airflow 技能吗？本课程基于 Airflow 3.2，带您掌握最新实践。您将使用 TaskFlow API 编写 Dags，借助 Assets 与全新的 Asset Partitions 基于数据进行调度，并通过重试、回调与测试提升稳定性。

在最后一章，您将基于 DuckDB 构建端到端的 SQL ETL 流水线，并直接在 Airflow 中加入数据质量检查，无需任何第三方库，确保流水线产出的数据始终可靠。完成课程后，您将掌握如何把原型流水线打造成可在生产环境中实际运行的方案。

Introduction to Apache Airflow in Python

使用TaskFlow API、基于资产的调度和可延迟传感器，运行带质量检查的SQL ETL管道。

使用 Airflow 构建数据流水线

使用 TaskFlow API、基于资产的调度和可延迟传感器编写 DAG，并运行带质量检查的端到端 SQL ETL 管道。



Airflow 的生产环境挑战

使用 Airflow 构建数据流水线

动手互动练习