首先，您将认识 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

from airflow.sdk import dag, task
from datetime import timedelta


@dag
def weather_pipeline():

    # Configure retry strategy
    @task(
        retries=____,
        retry_delay=timedelta(minutes=____),
        retry_exponential_backoff=____,
    )
    def fetch_weather():
        import random
        if random.random() < 0.3:
            raise Exception("API timeout")
        return {"temp": 22, "humidity": 65}

    @task
    def store_forecast(data):
        print(f"Stored forecast: temp={data['temp']}C, humidity={data['humidity']}%")

    weather = fetch_weather()
    store_forecast(weather)


pipeline = weather_pipeline()

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


from airflow.sdk import dag, task
from datetime import timedelta


@dag
def weather_pipeline():

    # Configure retry strategy
    @task(
        retries=3,
        retry_delay=timedelta(minutes=2),
        retry_exponential_backoff=2.0,
    )
    def fetch_weather():
        import random
        if random.random() < 0.3:
            raise Exception("API timeout")
        return {"temp": 22, "humidity": 65}

    @task
    def store_forecast(data):
        print(f"Stored forecast: temp={data['temp']}C, humidity={data['humidity']}%")

    weather = fetch_weather()
    store_forecast(weather)


pipeline = weather_pipeline()

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


使用指数退避配置重试

from airflow.sdk import dag, task


def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")


# Set the Dag to auto-pause after consecutive failures
@dag(max_consecutive_failed_dag_runs=____)
def monitoring_pipeline():

    # Attach the failure callback to the task
    @task(on_failure_callback=____)
    def check_api_health():
        raise Exception("API unreachable: connection refused")

    check_api_health()


pipeline = monitoring_pipeline()

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


from airflow.sdk import dag, task


def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")


# Set the Dag to auto-pause after consecutive failures
@dag(max_consecutive_failed_dag_runs=3)
def monitoring_pipeline():

    # Attach the failure callback to the task
    @task(on_failure_callback=alert_on_failure)
    def check_api_health():
        raise Exception("API unreachable: connection refused")

    check_api_health()


pipeline = monitoring_pipeline()

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


自动暂停失败的 Dag

选择合适的故障处理策略

可延迟运算符

from airflow.sdk import dag, task, chain
from airflow.providers.standard.sensors.filesystem import FileSensor


@dag
def file_processor():

    # Configure the sensor for production use
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/report.csv",
        mode="____",
        poke_interval=____,
        timeout=____,
    )

    @task
    def process_report():
        print("Processing incoming report")
        return "report_processed"

    chain(wait_for_data, process_report())


pipeline = file_processor()

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


from airflow.sdk import dag, task, chain
from airflow.providers.standard.sensors.filesystem import FileSensor


@dag
def file_processor():

    # Configure the sensor for production use
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/report.csv",
        mode="reschedule",
        poke_interval=30,
        timeout=300,
    )

    @task
    def process_report():
        print("Processing incoming report")
        return "report_processed"

    chain(wait_for_data, process_report())


pipeline = file_processor()

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


为生产环境配置传感器

from airflow.sdk import dag, task, chain
from airflow.providers.standard.sensors.filesystem import FileSensor


@dag
def deferred_file_processor():

    # Enable deferrable mode and set timeout
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/sales.csv",
        deferrable=____,
        timeout=____,
    )

    @task
    def process_sales():
        print("Processing sales data")
        return "sales_processed"

    chain(wait_for_data, process_sales())


pipeline = deferred_file_processor()

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


from airflow.sdk import dag, task, chain
from airflow.providers.standard.sensors.filesystem import FileSensor


@dag
def deferred_file_processor():

    # Enable deferrable mode and set timeout
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/sales.csv",
        deferrable=True,
        timeout=3600,
    )

    @task
    def process_sales():
        print("Processing sales data")
        return "sales_processed"

    chain(wait_for_data, process_sales())


pipeline = deferred_file_processor()

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


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

何时使用 poke、reschedule 或 deferrable

测试 Airflow 代码

import pytest
from airflow.models import DagBag

dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    # Assert that there are no import errors
    assert len(dag_bag.____) == 0, f"Import errors: {dag_bag.import_errors}"


def test_dag_loaded():
    # Assert the expected Dag is present
    assert "____" in dag_bag.____, "Expected Dag not found in DagBag"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


import pytest
from airflow.models import DagBag

dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    # Assert that there are no import errors
    assert len(dag_bag.import_errors) == 0, f"Import errors: {dag_bag.import_errors}"


def test_dag_loaded():
    # Assert the expected Dag is present
    assert "daily_etl" in dag_bag.dags, "Expected Dag not found in DagBag"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


编写 Dag 完整性测试

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pytest
from dags.data_cleaning import clean_record


def test_strips_whitespace():
    result = clean_record({"name": "  Alice  ", "email": "alice@test.com"})
    # Verify whitespace is stripped from the name
    assert result["name"] == "____", "Name should have whitespace stripped"


def test_lowercases_email():
    result = clean_record({"name": "Bob", "email": "BOB@Example.COM"})
    # Verify the email is lowercased
    assert result["email"] == "____", "Email should be lowercased"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


from airflow.sdk import dag, task


def clean_record(record):
    """Clean a single record by stripping whitespace and lowercasing email."""
    return {
        "name": record["name"].strip(),
        "email": record["email"].lower(),
    }


@dag
def cleaning_pipeline():

    @task
    def extract():
        return [
            {"name": "  Alice  ", "email": "ALICE@Example.COM"},
            {"name": "Bob  ", "email": "bob@EXAMPLE.com"},
        ]

    @task
    def transform(records):
        return [clean_record(r) for r in records]

    @task
    def load(cleaned):
        for r in cleaned:
            print(f"Loaded: {r['name']} ({r['email']})")

    raw = extract()
    cleaned = transform(raw)
    load(cleaned)


pipeline = cleaning_pipeline()

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


import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pytest
from dags.data_cleaning import clean_record


def test_strips_whitespace():
    result = clean_record({"name": "  Alice  ", "email": "alice@test.com"})
    # Verify whitespace is stripped from the name
    assert result["name"] == "Alice", "Name should have whitespace stripped"


def test_lowercases_email():
    result = clean_record({"name": "Bob", "email": "BOB@Example.COM"})
    # Verify the email is lowercased
    assert result["email"] == "bob@example.com", "Email should be lowercased"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


对任务函数进行单元测试

import json
from pathlib import Path
import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    # Load the Dag from the DagBag
    dag = dag_bag.get_dag("____")
    assert dag is not None, "Dag not found in DagBag"

    # Run the full pipeline as an integration test
    dag.____(logical_date=datetime(2026, 1, 15))

    # Verify the output file was created
    output = Path("____")
    assert output.exists(), "Pipeline did not produce output"

    # Verify the output contains the expected number of results
    results = json.loads(output.read_text())
    assert len(results) == ____, "Expected 2 results"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


import json
from pathlib import Path
import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    # Load the Dag from the DagBag
    dag = dag_bag.get_dag("etl_output")
    assert dag is not None, "Dag not found in DagBag"

    # Run the full pipeline as an integration test
    dag.test(logical_date=datetime(2026, 1, 15))

    # Verify the output file was created
    output = Path("/tmp/etl_results.json")
    assert output.exists(), "Pipeline did not produce output"

    # Verify the output contains the expected number of results
    results = json.loads(output.read_text())
    assert len(results) == 2, "Expected 2 results"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


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

使用 Task Groups 组织复杂的 Dags

# Import a task group
from airflow.sdk import dag, task, ____


@dag
def order_processing():

    # Set a custom group_id and default_args for the group
    @task_group(group_id="____", default_args={"retries": ____})
    def process_orders():

        @task
        def extract_orders():
            return [{"id": 1, "amount": 99.99}, {"id": 2, "amount": 45.50}]

        @task
        def transform_orders(orders):
            return [{"id": o["id"], "total": round(o["amount"] * 1.08, 2)} for o in orders]
          
		# Call extract and pass to transform
        raw = ____()
        return ____(raw)

    @task
    def load(results):
        print(f"Loaded {len(results)} orders to warehouse")

    orders = process_orders()
    load(orders)


pipeline = order_processing()

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


# Import a task group
from airflow.sdk import dag, task, task_group


@dag
def order_processing():

    # Set a custom group_id and default_args for the group
    @task_group(group_id="ingest_orders", default_args={"retries": 3})
    def process_orders():

        @task
        def extract_orders():
            return [{"id": 1, "amount": 99.99}, {"id": 2, "amount": 45.50}]

        @task
        def transform_orders(orders):
            return [{"id": o["id"], "total": round(o["amount"] * 1.08, 2)} for o in orders]
          
		# Call extract and pass to transform
        raw = extract_orders()
        return transform_orders(raw)

    @task
    def load(results):
        print(f"Loaded {len(results)} orders to warehouse")

    orders = process_orders()
    load(orders)


pipeline = order_processing()

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


将相关任务分组

from airflow.sdk import dag, task, task_group


@dag
def multi_stream_pipeline():

    # Set a display name for the parent group
    @task_group(group_display_name="____")
    def process_all():

        @task_group(group_display_name="Ingest Orders")
        def orders():
            @task
            def extract():
                return [{"id": 1, "amount": 99.99}]

            @task
            def transform(data):
                return [{"id": o["id"], "total": o["amount"] * 1.08} for o in data]

            return transform(extract())

        # Add a nested task group for returns with a display name
        @task_group(group_display_name="____")
        def returns():
            @task
            def extract():
                return [{"id": 1, "refund": 25.00}]

            @task
            def transform(data):
                return [{"id": r["id"], "credit": r["refund"] * -1} for r in data]

            return transform(extract())
          
		# Include both groups in the result
        return {"orders": orders(), "returns": ____()}

    @task
    def generate_report(data):
        print(f"Orders: {len(data['orders'])}, Returns: {len(data['returns'])}")

    results = process_all()
    generate_report(results)


pipeline = multi_stream_pipeline()

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


from airflow.sdk import dag, task, task_group


@dag
def multi_stream_pipeline():

    # Set a display name for the parent group
    @task_group(group_display_name="Process All")
    def process_all():

        @task_group(group_display_name="Ingest Orders")
        def orders():
            @task
            def extract():
                return [{"id": 1, "amount": 99.99}]

            @task
            def transform(data):
                return [{"id": o["id"], "total": o["amount"] * 1.08} for o in data]

            return transform(extract())

        # Add a nested task group for returns with a display name
        @task_group(group_display_name="Process Returns")
        def returns():
            @task
            def extract():
                return [{"id": 1, "refund": 25.00}]

            @task
            def transform(data):
                return [{"id": r["id"], "credit": r["refund"] * -1} for r in data]

            return transform(extract())
          
		# Include both groups in the result
        return {"orders": orders(), "returns": returns()}

    @task
    def generate_report(data):
        print(f"Orders: {len(data['orders'])}, Returns: {len(data['returns'])}")

    results = process_all()
    generate_report(results)


pipeline = multi_stream_pipeline()

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


使用可见名称嵌套任务组

选项

按领域

按算子类型

按领域组织任务

为生产环境准备 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 管道。



按领域组织任务

使用 Airflow 构建数据流水线

动手互动练习