编辑“︁Airflow TaskFlow API”︁（章节）

== 实际案例：数据ETL管道 ==

以下是一个完整的数据处理管道示例：

<syntaxhighlight lang="python">
from airflow.decorators import dag, task
from datetime import datetime
import pandas as pd

@dag(schedule="@daily", start_date=datetime(2023, 1, 1))
def sales_pipeline():

    @task(retries=3)
    def fetch_sales(date: datetime) -> pd.DataFrame:
        # 模拟从数据库获取数据
        return pd.DataFrame({
            "date": [date.date()],
            "revenue": [1000 * (date.day % 7 + 1)]
        })

    @task
    def calculate_stats(df: pd.DataFrame) -> dict:
        return {
            "total": df["revenue"].sum(),
            "average": df["revenue"].mean()
        }

    @task
    def store_report(stats: dict):
        print(f"Storing report: {stats}")

    sales_data = fetch_sales("{{ ds }}")
    stats = calculate_stats(sales_data)
    store_report(stats)

dag = sales_pipeline()
</syntaxhighlight>

'''关键点：'''
* 使用 Pandas DataFrame 作为数据类型
* 任务自动重试机制（`retries=3`）
* 模板变量 `{{ ds }}` 的用法
* 完整的类型安全数据流