Unknown-333/authoring-airflow-dags

编写 Airflow DAG

何时使用

• 创建或重构 Airflow DAG 和任务。
• 配置调度、回填、重试和 SLA。
• 在任务之间传递数据（XCom）或使用连接/变量。
• 不要用于诊断已崩溃的运行中 DAG（请使用 debugging-airflow-pipelines）。

工作流

- [ ] 使每个任务幂等，并由数据间隔参数化
- [ ] 将昂贵/导入密集的代码放在任务内部，而不是模块顶层
- [ ] 有意设置调度 + 回填
- [ ] 配置重试、重试延迟和 SLA
- [ ] 通过 TaskFlow 返回值或 >> 操作符编排依赖

1. 幂等任务 — 对于 2026-01-15 间隔的任务，无论运行一次还是重新运行，都必须产生相同的结果。使用数据间隔，而不是 datetime.now()。
2. 没有大量的顶层代码 — 调度器频繁解析每个 DAG 文件；数据库调用、API 调用或模块级别的大导入会拖慢调度，甚至导致解析失败。将它们放在任务内部。
3. 有目的地设置调度 + 回填 — catchup=True 会从 start_date 开始回填所有错过的间隔；默认设置为 False，除非你明确需要。
4. 重试和 SLA — 临时故障是正常的；设置 retries 和 retry_delay；使用 SLA/警报处理延迟。

模式

TaskFlow DAG，幂等且干净编排：

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["orders"],
)
def orders_pipeline():

    @task
    def extract(data_interval_start=None):
        # 使用间隔，而不是 now()，使重跑具有确定性。
        return fetch_orders(day=data_interval_start.date())

    @task
    def load(rows):
        # 删除-插入分区 -> 重试时幂等。
        overwrite_partition("fct_orders", rows)

    load(extract())

orders_pipeline()

通过 XCom 传递小数据（返回值）；通过存储传递大数据 — 写入 S3/GCS/数仓并传递路径/键，绝不对通过 XCom 传递兆字节。

使用连接/变量 处理密钥和配置（BaseHook.get_connection，Variable.get），绝不硬编码凭据。

常见陷阱

• 顶层的 API/数据库调用或大量导入 — 拖慢调度器，可能导致整个部署的 DAG 解析失败。
• 任务内部的 datetime.now() — 破坏幂等性和回填；请使用 data_interval_start/_end。
• 不经意地设置 catchup=True — 在首次部署时用历史运行充斥集群。
• 通过 XCom 传递大量负载 — 使元数据库膨胀；传递引用。
• 动态 start_date（例如 days_ago） — 使调度不确定；使用固定时间戳。
• 单一巨无霸任务 — 拆分提取/转换/加载，使重试粒度更细。

参考

• 调度、回填和反向填充参考