需要编写生产级 Apache Airflow DAG…

在创建、测试或调试 Airflow 数据管道与批处理调度 DAG 时，按生产级 DAG、operator、sensor 和测试模式落地。

适用场景

解决思路

1. 设计 DAG 时遵循四项原则：任务幂等（重复运行得到相同结果）、原子性（任务要么完全成功要么完全失败）、增量处理（只处理新增或变更数据）、可观测（每一步都有日志、指标和告警）。
2. 用依赖性语法表达任务关系：线性用 `task1 >> task2 >> task3`；扇出用 `task1 >> [task2, task3, task4]`；扇入用 `[task1, task2, task3] >> task4`；复杂依赖拆成多条链，如 `task1 >> task2 >> task4` 和 `task1 >> task3 >> task4`。
3. 按 Quick Start 在 `dags/example_dag.py` 中创建 DAG：导入 `datetime`、`timedelta`、`DAG`、`PythonOperator`、`EmptyOperator`。
4. 配置 `default_args`：`owner='data-team'`、`depends_on_past=False`、`email_on_failure=True`、`email_on_retry=False`、`retries=3`、`retry_delay=timedelta(minutes=5)`、`retry_exponential_backoff=True`、`max_retry_delay=timedelta(hours=1)`。
5. 定义 DAG 参数：`dag_id='example_etl'`、`default_args=default_args`、`description='Example ETL pipeline'`、`schedule='0 6 * * *'`、`start_date=datetime(2024, 1, 1)`、`catchup=False`、`tags=['etl', 'example']`、`max_active_runs=1`。
6. 创建 `start = EmptyOperator(task_id='start')`。
7. 实现 `extract_data(**context)`：读取 `execution_date = context['ds']`，执行提取逻辑并返回如 `{'records': 1000}`。
8. 创建 `extract = PythonOperator(task_id='extract', python_callable=extract_data)`。
9. 创建 `end = EmptyOperator(task_id='end')`。
10. 设置任务顺序：`start >> extract >> end`。
11. 使用 TaskFlow API：让代码更简洁，并自动处理 XCom。
12. 为任务设置 timeouts，防止出现 zombie tasks。
13. 对 sensors 使用 `mode='reschedule'`，释放 worker 槽位。
14. 为 DAG 编写单元测试和集成测试。
15. 保证任务幂等，使其可安全重试。
16. 不要使用 `depends_on_past=True`，避免产生瓶颈。
17. 不要硬编码日期，使用 `{{ ds }}` 宏。
18. 不要使用全局状态，任务应保持无状态。
19. 不要盲目跳过 catchup，先理解其影响。
20. 不要把重逻辑放在 DAG 文件中，从独立模块导入。
21. 详细模式和工作示例在 `references/details.md`；当上面的导航层不够用时读取该文件。

适用边界 · 注意事项

• 未使用 Apache Airflow 的项目。
• 只需简单 cron 定时脚本、没有复杂依赖、重试、回溯或监控需求的场景。
• 无法部署 Airflow 调度器、Metadata DB 和 Executor 的环境。