Airflow DAG Orchestrator

Airflow DAG Orchestrator 是做什么的？

设计并运营 Apache Airflow DAG，实现可靠的数据管道编排，包含依赖管理、SLA 和监控。

激活触发词

激活于： "Airflow"， "DAG"， "operator"， "sensor"， "scheduler"， "task dependency"， "SLA"， "backfill"， "XCom"， "TaskFlow API"， "MWAA"， "Cloud Composer"

不用于： dbt 模型执行 → dbt-analytics-engineer（尽管 Airflow 可以触发 dbt）| 流处理 → streaming-pipeline-architect | 工作流引擎（Temporal）→ distributed-transaction-manager

快速开始

1. 定义 DAG — 使用 TaskFlow API（@dag， @task 装饰器）编写 Python 原生 DAG
2. 设置调度 — cron 表达式或时间表，除非有意回填，否则设置 catchup=False
3. 配置重试 — 每个任务设置 retries=2， retry_delay=timedelta(minutes=5)
4. 添加 SLA — 在关键路径任务上设置 sla=timedelta(hours=2)
5. 本地测试 — 部署前运行 airflow dags test my_dag 2026-01-01

核心能力

领域	技术
Airflow	Apache Airflow 2.10+， MWAA， Cloud Composer 3
操作符	BashOperator， PythonOperator， KubernetesPodOperator
提供者	apache-airflow-providers-{snowflake, google, aws, dbt-cloud}
执行器	CeleryExecutor， KubernetesExecutor， LocalExecutor
监控	SLA 未命中， 任务耗时， Airflow 指标 → Prometheus

架构模式

TaskFlow API DAG

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

@dag(
    schedule="0 6 * * *",          # 每天 UTC 时间 6:00
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(hours=2),
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task()
    def extract_payments() -> dict:
        """从 Stripe API 提取数据"""
        data = stripe_client.list_payments(date=today())
        return {"count": len(data), "path": "s3://raw/payments/"}

    @task()
    def extract_orders() -> dict:
        """从 Shopify API 提取数据"""
        data = shopify_client.list_orders(date=today())
        return {"count": len(data), "path": "s3://raw/orders/"}

    @task()
    def transform(payments: dict, orders: dict) -> str:
        """在 DuckDB 中连接并转换"""
        result_path = run_duckdb_transform(payments["path"], orders["path"])
        return result_path

    @task()
    def load(path: str):
        """加载到 Snowflake"""
        snowflake_copy_into("fct_revenue", path)

    # 通过函数调用定义依赖关系
    payments = extract_payments()
    orders = extract_orders()
    transformed = transform(payments, orders)
    load(transformed)

daily_revenue_pipeline()

动态任务映射（扇出/扇入）

@task()
def get_partitions() -> list[str]:
    return ["2026-01-01", "2026-01-02", "2026-01-03"]

@task()
def process_partition(partition_date: str) -> dict:
    """每个分区并行执行"""
    return {"date": partition_date, "rows": process(partition_date)}

@task()
def aggregate(results: list[dict]):
    """扇入：接收所有分区结果"""
    total = sum(r["rows"] for r in results)
    log.info(f"Processed {total} total rows")

# 动态地将 process_partition 映射到所有分区
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)

dbt + Airflow 集成

from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# 选项 1：cosmos（推荐）
dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/airflow/dbt/"),
    profile_config=ProfileConfig(
        profile_name="default",
        target_name="prod",
    ),
    schedule="@daily",
    dag_id="dbt_daily",
)

# 选项 2：BashOperator（简单）
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)

反模式

1. 胖任务 — 任务应当是原子单元；不要将提取+转换+加载放在一个任务中
2. XCom 用于大数据 — XCom 用于元数据（路径、计数）；通过 S3/GCS 传递数据，而不是 XCom pickles
3. 意外使用 catchup=True — 除非需要回填，否则设置 catchup=False；否则 Airflow 会运行每个错过的间隔
4. 无重试 — 临时故障很常见；始终设置 retries >= 1 并带有延迟
5. DAG 文件中的顶级代码 — DAG 文件每 30 秒解析一次；模块级的大量导入或 API 调用会拖慢调度器

质量检查清单

• 使用 TaskFlow API 编写 Python 原生 DAG（而不是传统操作符链）
• 除非有意回填，否则设置 catchup=False
• 所有任务都配置了重试（至少 1 次重试）
• 在关键路径任务上设置了 SLA
• XCom 传递引用（S3 路径），而不是数据负载
• 使用动态任务映射实现扇出并行
• 部署前使用 airflow dags test 在本地测试 DAG
• 任务幂等性：重新运行任务产生相同结果
• 配置了任务失败和 SLA 未命中的告警
• 为 Airflow UI 中的过滤应用了 DAG 标签