Airflow dag builder

何时使用

• 需要用 Apache Airflow 编排数据管道：设计 DAG 结构与任务依赖、调度 ETL/批处理作业。
• 实现自定义 operator / sensor，等待外部文件、S3 对象或上游 DAG 完成。
• 本地测试 DAG（DagBag 加载校验、依赖断言）、排查失败的 DAG run。
• 触发词：airflow、DAG、数据管道、工作流编排、ETL 调度、批处理调度、TaskFlow、sensor、operator、cron 调度。

不该用（边界）：

• 实时流处理 / 毫秒级低延迟 → 用 Flink/Kafka Streams，Airflow 是批调度器，最小粒度是任务级。
• 非数据类工作流（如 CI/CD、前端构建）→ 用各自专用工具，别套 DAG。
• 单机一次性脚本、无依赖无调度需求 → 直接 cron + 脚本即可，别引入 Airflow。
• 任务内的具体 SQL 转换 / CSV 清洗逻辑本身 → 见「互见」，本技能只负责编排与调度。

步骤 / 指令

按序执行，每步可独立验证：

1. 确立 DAG 设计原则（四条硬约束）：
   • 幂等（Idempotent）：跑两次结果一致 —— 用 {{ ds }} 等模板取执行日期，禁止 datetime.now() 决定写入位置。
   • 原子（Atomic）：任务要么全成功要么全失败，便于安全重试。
   • 增量（Incremental）：只处理新增/变更数据，按 ds 分区。
   • 可观测（Observable）：每步有日志、指标、告警。
2. 建骨架 DAG：设 start_date、schedule（cron 或 @daily）、catchup=False、max_active_runs=1、tags；在 default_args 里配重试（retries=3、retry_delay、retry_exponential_backoff=True、max_retry_delay）。
3. 定义依赖：用位移运算符表达拓扑——
   • 线性 task1 >> task2 >> task3；
   • 扇出 task1 >> [task2, task3, task4]；
   • 扇入 [task1, task2, task3] >> task4。
4. 优先用 TaskFlow API（Airflow 2.0+）：@dag / @task 装饰器，函数返回值自动经 XCom 传递，代码更干净。
5. 按需选模式：分支用 BranchPythonOperator + join 任务（trigger_rule=NONE_FAILED_MIN_ONE_SUCCESS）；外部依赖用 sensor，务必 mode='reschedule' 释放 worker 槽位；批量同构管道用工厂函数 + globals() 动态生成 DAG。
6. 错误处理与告警：配 on_failure_callback（推 Slack/PagerDuty），清理任务用 trigger_rule=ALL_DONE（上游失败也跑），成功通知用 ALL_SUCCESS。
7. 测试 DAG：用 DagBag 断言 import_errors==0、任务数、调度、依赖关系、无环（test_cycle()）；任务内函数另写单元测试。
8. 组织项目结构：DAG 文件只放编排，重逻辑 import 自 dags/common/ 等模块；按 etl/、ml/ 分目录。

示例

最小可用 DAG（经典写法）：

# dags/example_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator

default_args = {
    'owner': 'data-team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(hours=1),
}

with DAG(
    dag_id='example_etl',
    default_args=default_args,
    schedule='0 6 * * *',          # 每天 6:00
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'example'],
    max_active_runs=1,
) as dag:
    start = EmptyOperator(task_id='start')

    def extract_data(**context):
        execution_date = context['ds']  # 用模板，保证幂等
        return {'records': 1000}

    extract = PythonOperator(task_id='extract', python_callable=extract_data)
    end = EmptyOperator(task_id='end')
    start >> extract >> end

TaskFlow API（XCom 自动传值）：

from airflow.decorators import dag, task

@dag(dag_id='taskflow_etl', schedule='@daily',
     start_date=datetime(2024, 1, 1), catchup=False, tags=['etl'])
def taskflow_etl():
    @task()
    def extract(source: str) -> dict: ...
    @task()
    def transform(extracted: dict) -> dict: ...
    @task()
    def load(transformed: dict, target: str): ...

    load(transform(extract('raw_data')), target='processed')

taskflow_etl()

Sensor 等外部依赖（mode='reschedule' 必加）：

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_file = S3KeySensor(
    task_id='wait_for_s3_file',
    bucket_name='data-lake',
    bucket_key='raw/{{ ds }}/data.parquet',
    timeout=60 * 60 * 2,    # 2 小时上限
    poke_interval=60 * 5,   # 每 5 分钟探一次
    mode='reschedule',      # 等待时让出 worker
)

DAG 测试：

import pytest
from airflow.models import DagBag

@pytest.fixture
def dagbag():
    return DagBag(dag_folder='dags/', include_examples=False)

def test_dag_loaded(dagbag):
    assert len(dagbag.import_errors) == 0, dagbag.import_errors

def test_dag_integrity(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.test_cycle() is None  # 无环

注意事项

应做（Do's）：

• 优先 TaskFlow API，代码更清晰、XCom 自动化。
• 给任务设超时（execution_timeout），防僵尸任务。
• sensor 一律 mode='reschedule'，长等待不占 worker。
• DAG 写单元测试 + 集成测试（DagBag 校验）。
• 任务保持幂等，可安全重试。

避免（Don'ts）：

• 别用 depends_on_past=True —— 制造串行瓶颈。
• 别硬编码日期 —— 用 {{ ds }} 等宏。
• 别用全局状态 —— 任务应无状态。
• 别盲目 catchup=False/True —— 先搞清回填的影响再设。
• 别把重逻辑写进 DAG 文件 —— DAG 文件会被 scheduler 反复解析，重逻辑要 import 自模块。

互见

• related：sql-query-builder — DAG 任务里执行的 SQL 转换/抽取语句由其生成。
• related：csv-data-cleaner — extract 后、load 前的表格清洗步骤可由其承担。
• related：rag-pipeline-builder — 用 Airflow 定时调度 RAG 语料的增量入库/重建管道。

---

本条采编自 wshobson/agents（MIT）。