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

Airflow 核心概念

编写最小化的 TaskFlow Dag

将 Airflow 组件与其角色匹配

TaskFlow API

将经典 Dag 转换为 TaskFlow

添加任务并连接依赖关系

在 Airflow UI 中探索一个管道

在任务之间共享数据

从 TaskFlow 函数返回数据

将经典算子连接到 TaskFlow 任务

理解 XCom 的限制

使用 TaskFlow 与 XCom 编写 Dags

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

动态任务映射

from airflow.sdk import dag, task


@dag
def api_processor():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint):
        print(f"Processing: {endpoint}")
        return f"{endpoint}_done"

    # Get the list of endpoints
    endpoints = ____()
    # Map over the list with expand
    process_endpoint.____(endpoint=____)


pipeline = api_processor()

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


from airflow.sdk import dag, task


@dag
def api_processor():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint):
        print(f"Processing: {endpoint}")
        return f"{endpoint}_done"

    # Get the list of endpoints
    endpoints = get_endpoints()
    # Map over the list with expand
    process_endpoint.expand(endpoint=endpoints)


pipeline = api_processor()

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


使用 expand 处理多个端点

from airflow.sdk import dag, task


@dag
def api_processor_v2():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint, base_url):
        print(f"Fetching: {base_url}/{endpoint}")
        return f"{endpoint}_done"

    endpoints = get_endpoints()
    # Fix the shared parameter and expand over endpoints
    process_endpoint.____(base_url="https://api.company.com/v2").____(endpoint=endpoints)


pipeline = api_processor_v2()

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


from airflow.sdk import dag, task


@dag
def api_processor_v2():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint, base_url):
        print(f"Fetching: {base_url}/{endpoint}")
        return f"{endpoint}_done"

    endpoints = get_endpoints()
    # Fix the shared parameter and expand over endpoints
    process_endpoint.partial(base_url="https://api.company.com/v2").expand(endpoint=endpoints)


pipeline = api_processor_v2()

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


用 partial 固定共享参数

使用 zip 配对输入

基于 Assets 的数据感知调度

from airflow.sdk import dag, task, Asset


# Define the asset
sales_data = ____("s3://data-lake/sales/daily.csv")

@dag
def sales_producer():

    @task(outlets=[sales_data])
    def write_sales_data():
        print("Writing daily sales data")
        return "sales_data_written"

    write_sales_data()


pipeline = sales_producer()

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


from airflow.sdk import dag, task, Asset


# Define the asset
sales_data = Asset("s3://data-lake/sales/daily.csv")

@dag
def sales_producer():

    @task(outlets=[sales_data])
    def write_sales_data():
        print("Writing daily sales data")
        return "sales_data_written"

    write_sales_data()


pipeline = sales_producer()

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


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

from airflow.sdk import dag, task, Asset


# Define the same asset as the producer
sales_data = ____("s3://data-lake/sales/daily.csv")

# Schedule the Dag on the asset
@dag(____=[sales_data])
def sales_consumer():

    @task
    def analyze_sales():
        print("Analyzing latest sales data")
        return "analysis_complete"

    analyze_sales()


pipeline = sales_consumer()

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


from airflow.sdk import dag, task, Asset


# Define the same asset as the producer
sales_data = Asset("s3://data-lake/sales/daily.csv")

# Schedule the Dag on the asset
@dag(schedule=[sales_data])
def sales_consumer():

    @task
    def analyze_sales():
        print("Analyzing latest sales data")
        return "analysis_complete"

    analyze_sales()


pipeline = sales_consumer()

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


在 Asset 上调度消费者 Dag

直观体验基于 Asset 的调度

模板化、幂等性与回填

from airflow.sdk import dag
from airflow.providers.standard.operators.bash import BashOperator


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

    # Replace hardcoded date with Jinja template
    export = BashOperator(
        task_id="export_data",
        bash_command="echo 'Exporting data for {{ ____ }}' > /tmp/export_{{ ____ }}.csv",
    )


pipeline = daily_export()

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


from airflow.sdk import dag
from airflow.providers.standard.operators.bash import BashOperator


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

    # Replace hardcoded date with Jinja template
    export = BashOperator(
        task_id="export_data",
        bash_command="echo 'Exporting data for {{ ds }}' > /tmp/export_{{ ds }}.csv",
    )


pipeline = daily_export()

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


添加日期感知的模板

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


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

    # Complete the WHERE clause to filter by logical date
    staged_rows = SQLExecuteQueryOperator(
        task_id="get_staged_rows",
        conn_id="duckdb_default",
        sql="SELECT * FROM staging WHERE ____ = '{{ ds }}'",
    )

    # Add a preoperator to delete existing rows before inserting
    load = SQLInsertRowsOperator(
        task_id="load_sales",
        conn_id="duckdb_default",
        table_name="sales",
        columns=["date", "product", "amount"],
        preoperator="____ FROM sales WHERE date = '{{ ds }}';",
        rows=staged_rows.output,
    )


pipeline = idempotent_load()

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


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


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

    # Complete the WHERE clause to filter by logical date
    staged_rows = SQLExecuteQueryOperator(
        task_id="get_staged_rows",
        conn_id="duckdb_default",
        sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
    )

    # Add a preoperator to delete existing rows before inserting
    load = SQLInsertRowsOperator(
        task_id="load_sales",
        conn_id="duckdb_default",
        table_name="sales",
        columns=["date", "product", "amount"],
        preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
        rows=staged_rows.output,
    )


pipeline = idempotent_load()

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


让管道具备幂等性

触发回填

Human-in-the-loop 工作流

from airflow.sdk import dag, task, chain, Param
# Import HITLEntryOperator
from airflow.providers.standard.operators.____ import ____


@dag
def ticket_review():

    @task
    def generate_draft():
        # In production, this would call an LLM API to generate the draft
        return (
            "Hi Sarah, thanks for reaching out. I can see your last "
            "invoice was charged twice due to a billing system error. "
            "We have initiated a refund of $49.99 which should appear "
            "within 3-5 business days. Let me know if you need anything else."
        )

    # Complete the HITLEntryOperator with params
    review = ____(
        task_id="human_review",
        subject="Review AI draft response",
        body="{{ ti.xcom_pull(task_ids='generate_draft') }}",
        params={
            "____": ____("", type="string"),
            "____": ____("p3", type="string"),
        },
    )

    @task
    def send_response(hitl_output):
        feedback = hitl_output["params_input"]["feedback"]
        urgency = hitl_output["params_input"]["urgency"]
        print(f"Sending response (urgency: {urgency}): {feedback}")

    draft = generate_draft()
    chain(draft, review)
    send_response(review.output)


pipeline = ticket_review()

if __name__ == "__main__":
    from unittest.mock import patch

    mock_hitl_response = {
        "chosen_options": [],
        "params_input": {"feedback": "Looks good", "urgency": "p3"},
    }

    with patch(
        "airflow.providers.standard.operators.hitl.HITLEntryOperator.execute",
        return_value=mock_hitl_response,
    ):
        pipeline.test()


from airflow.sdk import dag, task, chain, Param
# Import HITLEntryOperator
from airflow.providers.standard.operators.hitl import HITLEntryOperator


@dag
def ticket_review():

    @task
    def generate_draft():
        # In production, this would call an LLM API to generate the draft
        return (
            "Hi Sarah, thanks for reaching out. I can see your last "
            "invoice was charged twice due to a billing system error. "
            "We have initiated a refund of $49.99 which should appear "
            "within 3-5 business days. Let me know if you need anything else."
        )

    # Complete the HITLEntryOperator with params
    review = HITLEntryOperator(
        task_id="human_review",
        subject="Review AI draft response",
        body="{{ ti.xcom_pull(task_ids='generate_draft') }}",
        params={
            "feedback": Param("", type="string"),
            "urgency": Param("p3", type="string"),
        },
    )

    @task
    def send_response(hitl_output):
        feedback = hitl_output["params_input"]["feedback"]
        urgency = hitl_output["params_input"]["urgency"]
        print(f"Sending response (urgency: {urgency}): {feedback}")

    draft = generate_draft()
    chain(draft, review)
    send_response(review.output)


pipeline = ticket_review()

if __name__ == "__main__":
    from unittest.mock import patch

    mock_hitl_response = {
        "chosen_options": [],
        "params_input": {"feedback": "Looks good", "urgency": "p3"},
    }

    with patch(
        "airflow.providers.standard.operators.hitl.HITLEntryOperator.execute",
        return_value=mock_hitl_response,
    ):
        pipeline.test()


使用 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 工作负载

使用 SQL 汇总销售数据

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

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

使用 Asset Partitions 的分区感知调度

创建分区管道

在分区资产上进行调度

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

数据质量检查

添加列级质量检查

添加表级数据质量检查

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

从开发到生产

从 CLI 触发一条流水线

回填历史数据

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 管道。



基于 Assets 的数据感知调度

Create Your Free Account