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

Airflow 核心概念

# Import the Dag and task decorators
from airflow.sdk import ____, ____


# Apply the Dag decorator
@____
def my_first_pipeline():

    # Apply the task decorator and return the expected string
    @____
    def start():
        ____ "Pipeline started"

    start()


pipeline = my_first_pipeline()

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


# Import the Dag and task decorators
from airflow.sdk import dag, task


# Apply the Dag decorator
@dag
def my_first_pipeline():

    # Apply the task decorator and return the expected string
    @task
    def start():
        return "Pipeline started"

    start()


pipeline = my_first_pipeline()

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


编写最小化的 TaskFlow Dag

Options

Orchestration

Execution

Interface & Storage

将 Airflow 组件与其角色匹配

TaskFlow API

from airflow.sdk import dag, task

# Add the Dag decorator
@____
def etl_pipeline():

    # Add the task decorator
    @____
    def extract_data():
        return {"users": 150, "events": 4200}

    # Add the task decorator
    @____
    def print_summary(data):
        print(f"Users: {data['users']}, Events: {data['events']}")

    # Wire dependencies by passing return values
    data = ____()
    ____(data)


pipeline = etl_pipeline()

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


# Classic version (read-only reference)
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator


def extract_data():
    return {"users": 150, "events": 4200}


def print_summary(data):
    print(f"Users: {data['users']}, Events: {data['events']}")


with DAG(dag_id="etl_pipeline") as dag:
    extract_task = PythonOperator(
        task_id="extract_data",
        python_callable=extract_data,
    )
    summary_task = PythonOperator(
        task_id="print_summary",
        python_callable=print_summary,
        op_args=[extract_task.output],
    )
    extract_task >> summary_task


from airflow.sdk import dag, task

# Add the Dag decorator
@dag
def etl_pipeline():

    # Add the task decorator
    @task
    def extract_data():
        return {"users": 150, "events": 4200}

    # Add the task decorator
    @task
    def print_summary(data):
        print(f"Users: {data['users']}, Events: {data['events']}")

    # Wire dependencies by passing return values
    data = extract_data()
    print_summary(data)


pipeline = etl_pipeline()

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

将经典 Dag 转换为 TaskFlow

from airflow.sdk import dag, task

@dag
def user_pipeline():

    @task
    def extract():
        return [
            {"id": 1, "name": "alice"},
            {"id": 2, "name": "bob"},
            {"id": 3, "name": "charlie"},
        ]

    @task
    def transform(raw_data):
        return [{"id": r["id"], "name": r["name"].upper()} for r in raw_data]

    # Add a decorated task function
    @____
    def load(____):
        print(f"Loaded {len(____)} records")

    raw = extract()
    cleaned = transform(raw)
    # Call the new task
    ____(cleaned)


pipeline = user_pipeline()

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


from airflow.sdk import dag, task

@dag
def user_pipeline():

    @task
    def extract():
        return [
            {"id": 1, "name": "alice"},
            {"id": 2, "name": "bob"},
            {"id": 3, "name": "charlie"},
        ]

    @task
    def transform(raw_data):
        return [{"id": r["id"], "name": r["name"].upper()} for r in raw_data]

    # Add a decorated task function
    @task
    def load(records):
        print(f"Loaded {len(records)} records")

    raw = extract()
    cleaned = transform(raw)
    # Call the new task
    load(cleaned)


pipeline = user_pipeline()

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


添加任务并连接依赖关系

在 Airflow UI 中探索一个管道

在任务之间共享数据

from airflow.sdk import dag, task

@dag
def xcom_example():

    @task
    def get_config():
        return {"pipeline_name": "daily_etl", "batch_size": 1000}

    @task
    def run_pipeline(config):
        print(f"Running {config['pipeline_name']}")

    # Call the config function and store the result
    config = ____()
    # Pass the variable to wire the dependency
    ____(config)


pipeline = xcom_example()

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


from airflow.sdk import dag, task

@dag
def xcom_example():

    @task
    def get_config():
        return {"pipeline_name": "daily_etl", "batch_size": 1000}

    @task
    def run_pipeline(config):
        print(f"Running {config['pipeline_name']}")

    # Call the config function and store the result
    config = get_config()
    # Pass the variable to wire the dependency
    run_pipeline(config)

pipeline = xcom_example()

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


从 TaskFlow 函数返回数据

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

@dag
def mixed_pipeline():

    get_timestamp = BashOperator(
        task_id="get_timestamp",
        bash_command="date +%Y-%m-%d",
    )

    @task
    def log_timestamp(ts_value):
        print(f"Timestamp received: {ts_value}")

    # Access the classic operator's result
    ts = get_timestamp.____
    # Pass the output to wire the dependency
    log_timestamp(____)


pipeline = mixed_pipeline()

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


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

@dag
def mixed_pipeline():

    get_timestamp = BashOperator(
        task_id="get_timestamp",
        bash_command="date +%Y-%m-%d",
    )

    @task
    def log_timestamp(ts_value):
        print(f"Timestamp received: {ts_value}")

    # Access the classic operator's result
    ts = get_timestamp.output
    # Pass the output to wire the dependency
    log_timestamp(ts)


pipeline = mixed_pipeline()

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


将经典算子连接到 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 工作负载

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


将经典 Dag 转换为 TaskFlow

使用 Airflow 构建数据流水线

动手互动练习