Apache Airflow DAG 生产级设计模式实战指南：基于 agents 数据工程技能构建可靠的编排管道

本指南围绕 agents 插件仓库中 data-engineering 插件的 airflow-dag-patterns 技能展开，系统讲解如何在 Apache Airflow 中设计、编排、测试与部署生产级 DAG。无论你是在搭建数据管道、设计任务依赖、实现自定义 Operator/Sensor，还是排查 DAG 运行失败，读完本文即可获得一套可直接落地的 DAG 骨架与六大进阶模式，并能将该技能的能力与仓库内 data-engineer 智能体 的调度知识相互印证。

技能定位：airflow-dag-patterns 在 agents 仓库中的位置

airflow-dag-patterns 是 data-engineering 插件下四大技能之一，与 spark-optimization、dbt-transformation-patterns、data-quality-frameworks 共同覆盖数据工程的调度编排、Spark 优化、dbt 转换与数据质量四大场景。其技能清单在 agent-skills.md 中登记为“Design Airflow DAGs with proper dependencies and error handling”，即面向 DAG 结构与依赖设计、错误处理的生产级 Airflow 编排指导。

从仓库目录结构可以直观看到该技能采用的“渐进式披露”（Progressive Disclosure）组织方式，与仓库对全部 183 个技能的规范一致：

plugins/data-engineering/skills/airflow-dag-patterns/
├── SKILL.md                 # YAML Frontmatter（元数据）+ 核心指引
└── references/
    └── details.md           # 六大模式与完整可运行示例（按需加载）

其中 SKILL.md 承担“元数据 + 指令”层：YAML Frontmatter 声明技能名称与激活条件（Use when 触发子句），正文给出 DAG 设计原则、依赖写法、Quick Start 与 Do's/Don'ts；而更细的“Detailed patterns and worked examples”被放进 references/details.md，仅在导航层不足以覆盖需求时才读取，从而在不牺牲信息密度的前提下节省上下文 Token。

在实际使用中，该技能与仓库内的编排类命令相辅相成：data-pipeline.md 负责数据管道整体架构设计（含 XCom、Task Group、重试策略等编排要点），而本技能则在“具体 DAG 怎么写、模式怎么选、如何测试”的粒度上给出可执行范式；data-engineer 智能体 的能力描述也明确列出 “Apache Airflow with custom operators and dynamic DAG generation”，正是本技能两类核心模式的应用场景。

触发时机：何时使用该技能

依据 SKILL.md 的说明，以下场景应当激活本技能：

• 使用 Airflow 构建数据管道编排（Creating data pipeline orchestration with Airflow）
• 设计 DAG 结构与任务依赖（Designing DAG structures and dependencies）
• 实现自定义 Operator 与 Sensor（Implementing custom operators and sensors）
• 在本地测试 Airflow DAG（Testing Airflow DAGs locally）
• 生产环境部署 Airflow（Setting up Airflow in production）
• 排查失败的 DAG 运行（Debugging failed DAG runs）

核心概念一：DAG 设计的四大原则

任何生产级 DAG 都应围绕四个可验证的设计维度展开，SKILL.md 用一张表给出了定义，其工程含义如下：

原则（Principle）	含义	落地要点
Idempotent（幂等）	运行两次产生相同结果	写任务逻辑时保证重复执行不会重复插入/重复计费，是重试与补数据的前提
Atomic（原子性）	任务要么整体成功、要么整体失败	单个 Task 职责单一，避免“半成功”状态污染下游
Incremental（增量）	只处理新增/变化的数据	用 {{ ds }} 等宏锚定时间分区，配合 watermark 列做增量抽取
Observable（可观测）	每一步都有日志、指标与告警	结合回调（callback）、日志 URL、结构化告警形成闭环

这四项原则同样呼应 data-pipeline.md 中“记录数、失败数、执行耗时、成功率需被监控”的要求——只有把幂等、增量做进 DAG 设计，下游的数据质量与可观测性才有根基。

核心概念二：Task Dependencies 任务依赖

Airflow 用位运算操作符（>>、<<）在 Task 之间声明依赖，无需显式写顺序代码。SKILL.md 给出的四类基础形态如下：

# Linear 线性
task1 >> task2 >> task3

# Fan-out 扇出：task1 完成后并行执行三个分支
task1 >> [task2, task3, task4]

# Fan-in 汇聚：三个上游全部成功后执行 task4
[task1, task2, task3] >> task4

# Complex 复杂组合：task2、task3 可并行，task4 等待二者
task1 >> task2 >> task4
task1 >> task3 >> task4

理解要点：

• a >> b 等价于 a.set_downstream(b)，即“a 先于 b”；a << b 为反向“b 先于 a”。
• 列表与单个 Operator 混用（如 [task1, task2] >> task4）表示 fan-in；单个 Operator 接列表表示 fan-out。
• 分支内任务并行度由调度器依据依赖图与执行器并发上限决定，DAG 作者只需要声明“谁依赖谁”。
• 无论拓扑多复杂，Airflow 会按依赖关系计算执行顺序；这也是 details.md 中“测试 DAG 无环”的验证基础——dag.test_cycle() 即用来检测依赖图中是否出现循环。

Quick Start：一个可直接运行的 ETL 示例

SKILL.md 给出了一个完整的入门 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',
    '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),
}

with DAG(
    dag_id='example_etl',
    default_args=default_args,
    description='Example ETL pipeline',
    schedule='0 6 * * *',  # Daily at 6 AM
    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']
        # Extract logic here
        return {'records': 1000}

    extract = PythonOperator(
        task_id='extract',
        python_callable=extract_data,
    )

    end = EmptyOperator(task_id='end')

    start >> extract >> end

default_args 参数详解

default_args 中的配置会被组内所有 Task 继承（Task 自身显式声明的同名参数优先级更高），理解每个键是写出可靠重试策略的前提：

参数	作用与建议
owner	负责人标识，便于告警与追溯归属
depends_on_past	是否依赖上一调度周期成功；生产中默认 False，置 True 会形成串联瓶颈
email_on_failure	任务失败时是否发邮件告警
email_on_retry	重试时是否发邮件；通常 False 避免告警风暴
retries	失败后的自动重试次数（如 3）
retry_delay	每次重试前的等待间隔（如 5 分钟）
retry_exponential_backoff	是否指数退避：等待时间随重试次数指数增长，降低对下游系统的冲击
max_retry_delay	指数退避的等待上限（如 1 小时），防止退避时间无限拉长

DAG 级参数说明

• dag_id：全局唯一标识，会出现在 UI、日志、告警与 API 中。
• schedule：调度表达式。支持 cron（'0 6 * * *' 表示每天 06:00）与预设（@daily、@hourly、@weekly 等）。Airflow 2.4+ 推荐用 schedule 替代旧的 schedule_interval。
• start_date：DAG 的起始日期，配合 catchup 决定补跑行为。
• catchup=False：不追溯补跑启动前错过的周期；盲目开启追跑（backfill）在数据量大时会瞬时打爆调度队列。
• tags：给 DAG 打标签，便于 UI 筛选（如 etl、example）。
• max_active_runs=1：同一时间只允许一个 DAG Run 在跑，避免上一周期未结束、下一周期又启动造成的并发写冲突。

任务上下文与 ds

python_callable 函数声明 **context 后即可拿到 Airflow 注入的运行上下文。示例中的 context['ds'] 即“逻辑执行日期”的 YYYY-MM-DD 字符串，对应模板宏 {{ ds }}。不要在 DAG 代码里硬编码日期，一律通过 ds、ts、execution_date 等上下文或模板宏派生，这既是增量处理的锚点，也是数据回溯（rerun）可复现的前提。返回值的字典（如 {'records': 1000}）会被自动写入 XCom，供下游任务读取。

六大进阶模式与完整示例

当 SKILL.md 的导航层不足以覆盖场景时，应按 details.md 的指引读取深度示例。六个模式构成一条从“编码风格”到“生产运维”的完整能力链：TaskFlow API → 动态 DAG → 分支 → 传感器 → 错误处理与告警 → 测试。

模式一：TaskFlow API（Airflow 2.0+）

传统 PythonOperator 需要手写 op_kwargs、xcom_push/xcom_pull，样板代码多。TaskFlow API 用 @dag 与 @task 装饰器把普通 Python 函数变成 DAG 与任务，函数返回值自动通过 XCom 在任务间传递，只需把上游任务的返回值当作参数传入下游函数即可：

# dags/taskflow_example.py
from datetime import datetime
from airflow.decorators import dag, task
from airflow.models import Variable

@dag(
    dag_id='taskflow_etl',
    schedule='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'taskflow'],
)
def taskflow_etl():
    """ETL pipeline using TaskFlow API"""

    @task()
    def extract(source: str) -> dict:
        """Extract data from source"""
        import pandas as pd

        df = pd.read_csv(f's3://bucket/{source}/{{ ds }}.csv')
        return {'data': df.to_dict(), 'rows': len(df)}

    @task()
    def transform(extracted: dict) -> dict:
        """Transform extracted data"""
        import pandas as pd

        df = pd.DataFrame(extracted['data'])
        df['processed_at'] = datetime.now()
        df = df.dropna()
        return {'data': df.to_dict(), 'rows': len(df)}

    @task()
    def load(transformed: dict, target: str):
        """Load data to target"""
        import pandas as pd

        df = pd.DataFrame(transformed['data'])
        df.to_parquet(f's3://bucket/{target}/{{ ds }}.parquet')
        return transformed['rows']

    @task()
    def notify(rows_loaded: int):
        """Send notification"""
        print(f'Loaded {rows_loaded} rows')

    # Define dependencies with XCom passing
    extracted = extract(source='raw_data')
    transformed = transform(extracted)
    loaded = load(transformed, target='processed_data')
    notify(loaded)

# Instantiate the DAG
taskflow_etl()

使用要点：

• 依赖即“函数调用”：transform(extracted) 就同时完成了数据传递与依赖声明，notify(loaded) 前必须完成 load，XCom 的序列化/反序列化全部自动完成。
• 文件路径中 {{ ds }} 必须写成 f-string 双花括号转义（f's3://.../{{ ds }}.csv'），否则会在 DAG 解析期被提前渲染。
• 由于依赖来自函数参数，函数签名类型提示（-> dict、-> int）尽量写全，可读性与 IDE 支持更好。
• 该风格对应 data-pipeline.md 中“XCom for inter-task communication”的编排建议，是 Airflow 2.x 的首选写法。

模式二：动态 DAG 生成（工厂函数）

当多条管道“结构相同、参数不同”（如不同业务表的 ETL）时，不应复制粘贴 DAG，而应使用 DAG Factory 模式：用配置列表 + 工厂函数批量生成同名异参的 DAG，这正对应 data-engineer 智能体 提到的 “dynamic DAG generation” 能力。

# dags/dynamic_dag_factory.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
import json

# Configuration for multiple similar pipelines
PIPELINE_CONFIGS = [
    {'name': 'customers', 'schedule': '@daily', 'source': 's3://raw/customers'},
    {'name': 'orders', 'schedule': '@hourly', 'source': 's3://raw/orders'},
    {'name': 'products', 'schedule': '@weekly', 'source': 's3://raw/products'},
]

def create_dag(config: dict) -> DAG:
    """Factory function to create DAGs from config"""

    dag_id = f"etl_{config['name']}"

    default_args = {
        'owner': 'data-team',
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
    }

    dag = DAG(
        dag_id=dag_id,
        default_args=default_args,
        schedule=config['schedule'],
        start_date=datetime(2024, 1, 1),
        catchup=False,
        tags=['etl', 'dynamic', config['name']],
    )

    with dag:
        def extract_fn(source, **context):
            print(f"Extracting from {source} for {context['ds']}")

        def transform_fn(**context):
            print(f"Transforming data for {context['ds']}")

        def load_fn(table_name, **context):
            print(f"Loading to {table_name} for {context['ds']}")

        extract = PythonOperator(
            task_id='extract',
            python_callable=extract_fn,
            op_kwargs={'source': config['source']},
        )

        transform = PythonOperator(
            task_id='transform',
            python_callable=transform_fn,
        )

        load = PythonOperator(
            task_id='load',
            python_callable=load_fn,
            op_kwargs={'table_name': config['name']},
        )

        extract >> transform >> load

    return dag

# Generate DAGs
for config in PIPELINE_CONFIGS:
    globals()[f"dag_{config['name']}"] = create_dag(config)

要点与陷阱：

• 每个 config 会生成一个独立 DAG，dag_id 如 etl_customers、etl_orders、etl_products，调度频率互不影响（@daily/@hourly/@weekly）。
• 不同管道差异化的参数（source、table_name）通过 op_kwargs 注入对应任务，闭包/工厂负责隔离，避免共享可变状态。
• 关键陷阱：Airflow Scheduler 会周期性地重新解析整个 DAG 文件以发现变化。因此配置文件若写死为模块级常量（如示例的 PIPELINE_CONFIGS），新增管道需重发代码；若配置来自外部源（数据库、Variable、文件），则可在调度重解析时动态增删 DAG，这正是该模式适合“配置驱动平台”的原因。本示例显式 import 了 Variable 与 json，即暗示配置可演进为运行时来源。
• globals() 赋值必须在模块顶层执行，且要在文件末尾调用工厂，保证 DAG 对象在解析结束时已绑定。

模式三：分支逻辑与 TriggerRule

数据管道往往需要按数据特征走不同分支（如按质量分路由）。BranchPythonOperator 返回下游任务 id 决定走哪条路径；由于分支导致部分上游未执行，汇合点必须改用合适的 TriggerRule，否则默认的 all_success 会让汇合任务永远失败：

# dags/branching_example.py
from airflow.decorators import dag, task
from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule

@dag(
    dag_id='branching_pipeline',
    schedule='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
)
def branching_pipeline():

    @task()
    def check_data_quality() -> dict:
        """Check data quality and return metrics"""
        quality_score = 0.95  # Simulated
        return {'score': quality_score, 'rows': 10000}

    def choose_branch(**context) -> str:
        """Determine which branch to execute"""
        ti = context['ti']
        metrics = ti.xcom_pull(task_ids='check_data_quality')

        if metrics['score'] >= 0.9:
            return 'high_quality_path'
        elif metrics['score'] >= 0.7:
            return 'medium_quality_path'
        else:
            return 'low_quality_path'

    quality_check = check_data_quality()

    branch = BranchPythonOperator(
        task_id='branch',
        python_callable=choose_branch,
    )

    high_quality = EmptyOperator(task_id='high_quality_path')
    medium_quality = EmptyOperator(task_id='medium_quality_path')
    low_quality = EmptyOperator(task_id='low_quality_path')

    # Join point - runs after any branch completes
    join = EmptyOperator(
        task_id='join',
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
    )

    quality_check >> branch >> [high_quality, medium_quality, low_quality] >> join

branching_pipeline()

运行机制拆解：

1. check_data_quality（TaskFlow）返回质量分，自动写入 XCom。
2. choose_branch 通过 context['ti'].xcom_pull(task_ids='check_data_quality') 读回指标并返回分支任务 id；本例设了三个阈值档位：>=0.9、>=0.7、其余，分别对应 high/medium/low_quality_path。
3. 分支各自是一个独立任务；未被选中的分支任务会被标记为 skipped。
4. join 使用 TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS（只要有至少一个上游成功、且没有失败即触发），确保无论走哪条分支都能继续，这正是分支汇合的经典写法。其它常见规则还包括 ALL_DONE（无论成败都执行）、ALL_SUCCESS（全部成功才执行）、ONE_SUCCESS（任一成功即执行）。

模式四：Sensors 与外部依赖

Sensor 是“等待外部条件满足”的特殊任务。三类典型场景被组合在一个 DAG 中演示：等待 S3 文件就绪、等待上游 DAG 完成、轮询外部 API 健康状态：

# dags/sensor_patterns.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.python import PythonOperator

with DAG(
    dag_id='sensor_example',
    schedule='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    # Wait for file on S3
    wait_for_file = S3KeySensor(
        task_id='wait_for_s3_file',
        bucket_name='data-lake',
        bucket_key='raw/{{ ds }}/data.parquet',
        aws_conn_id='aws_default',
        timeout=60 * 60 * 2,  # 2 hours
        poke_interval=60 * 5,  # Check every 5 minutes
        mode='reschedule',  # Free up worker slot while waiting
    )

    # Wait for another DAG to complete
    wait_for_upstream = ExternalTaskSensor(
        task_id='wait_for_upstream_dag',
        external_dag_id='upstream_etl',
        external_task_id='final_task',
        execution_date_fn=lambda dt: dt,  # Same execution date
        timeout=60 * 60 * 3,
        mode='reschedule',
    )

    # Custom sensor using @task.sensor decorator
    @task.sensor(poke_interval=60, timeout=3600, mode='reschedule')
    def wait_for_api() -> PokeReturnValue:
        """Custom sensor for API availability"""
        import requests

        response = requests.get('https://api.example.com/health')
        is_done = response.status_code == 200

        return PokeReturnValue(is_done=is_done, xcom_value=response.json())

    api_ready = wait_for_api()

    def process_data(**context):
        api_result = context['ti'].xcom_pull(task_ids='wait_for_api')
        print(f"API returned: {api_result}")

    process = PythonOperator(
        task_id='process',
        python_callable=process_data,
    )

    [wait_for_file, wait_for_upstream, api_ready] >> process

关键参数与选择依据：

场景	使用	关键参数
数据文件到达（S3/HDFS/本地）	S3KeySensor（AWS 需安装 apache-airflow-providers-amazon）、FileSensor	bucket_name、bucket_key（支持 {{ ds }} 宏）、aws_conn_id
等待另一个 DAG 的某个 Task	ExternalTaskSensor	external_dag_id、external_task_id、execution_date_fn（对齐执行日期）
自定义轮询逻辑（API 健康等）	@task.sensor 装饰器	poke_interval、timeout、mode

• 轮询频率：poke_interval（默认约 60 秒）控制每次探测间隔，这里 S3 探测设为 5 分钟；timeout 为最长等待，超时任务即失败，本例分别为 2 小时与 3 小时。
• mode='reschedule' 是生产关键：与默认的 poke 模式在 Task 槽位内死等不同，reschedule 模式在两次探测之间会释放 worker 槽位，避免长时间等待的 Sensor 占满执行器——这正是 SKILL.md Best Practices 中“sensor 使用 mode='reschedule' 以释放 worker”的出处。注意 reschedule 模式对任务有额外约束（如超时后任务进入 deferred 状态并由 scheduler 重排）。
• @task.sensor 进阶能力：把自定义轮询逻辑写成装饰器函数，返回 PokeReturnValue(is_done=..., xcom_value=...)，既做条件判断又向 XCom 写入探测结果（如 API 返回的 JSON）。示例为简洁起见省略了该类型的导入，实际使用需补 from airflow.sensors.base import PokeReturnValue。
• 三个传感器 fan-in 汇合到 process，即“文件就绪 + 上游完成 + API 可达”三者全部满足才开始处理；process 用 context['ti'].xcom_pull(task_ids='wait_for_api') 读回 API 结果。

模式五：错误处理与告警

生产 DAG 必须有“失败可感知、清理可执行、成功可通知”的能力。模式五同时演示了**回调（callback）与触发规则（TriggerRule）**两个机制：

# dags/error_handling.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.models import Variable

def task_failure_callback(context):
    """Callback on task failure"""
    task_instance = context['task_instance']
    exception = context.get('exception')

    # Send to Slack/PagerDuty/etc
    message = f"""
    Task Failed!
    DAG: {task_instance.dag_id}
    Task: {task_instance.task_id}
    Execution Date: {context['ds']}
    Error: {exception}
    Log URL: {task_instance.log_url}
    """
    # send_slack_alert(message)
    print(message)

def dag_failure_callback(context):
    """Callback on DAG failure"""
    # Aggregate failures, send summary
    pass

with DAG(
    dag_id='error_handling_example',
    schedule='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    on_failure_callback=dag_failure_callback,
    default_args={
        'on_failure_callback': task_failure_callback,
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
    },
) as dag:

    def might_fail(**context):
        import random
        if random.random() < 0.3:
            raise ValueError("Random failure!")
        return "Success"

    risky_task = PythonOperator(
        task_id='risky_task',
        python_callable=might_fail,
    )

    def cleanup(**context):
        """Cleanup runs regardless of upstream failures"""
        print("Cleaning up...")

    cleanup_task = PythonOperator(
        task_id='cleanup',
        python_callable=cleanup,
        trigger_rule=TriggerRule.ALL_DONE,  # Run even if upstream fails
    )

    def notify_success(**context):
        """Only runs if all upstream succeeded"""
        print("All tasks succeeded!")

    success_notification = PythonOperator(
        task_id='notify_success',
        python_callable=notify_success,
        trigger_rule=TriggerRule.ALL_SUCCESS,
    )

    risky_task >> [cleanup_task, success_notification]

结构分析：

• 两级回调分工：on_failure_callback（经 default_args 注入所有 Task）负责单任务粒度的失败告警，组装 dag_id、task_id、ds、异常对象与 log_url 发送到 Slack/PagerDuty；dag_failure_callback 在 DAG 级汇总失败并生成摘要。回调收到的 context 是标准执行上下文，context['task_instance'].log_url 可直接定位失败日志。
• “无论成败都清理”：cleanup_task 用 TriggerRule.ALL_DONE，risky_task 无论成功还是抛异常都会触发清理，适合释放锁、关闭连接等收尾动作。
• “全成功才通知”：success_notification 用默认的 TriggerRule.ALL_SUCCESS，只有 risky_task 成功才发送成功通知。
• 该示例还与 SKILL.md 的重试体系串联：retries: 3 + retry_delay: 5min 保证瞬态失败自动恢复，callback 只在最终失败时报警，符合“可观测原则”的告警要求。

模式六：测试 DAG

“先写测试、再上生产”在 Airflow 中同样成立。details.md 给出了两类测试：结构测试（用 DagBag 验证 DAG 能被正确解析、依赖无环）与逻辑单元测试（直接调用 Python 函数验证业务逻辑）：

# tests/test_dags.py
import pytest
from datetime import datetime
from airflow.models import DagBag

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

def test_dag_loaded(dagbag):
    """Test that all DAGs load without errors"""
    assert len(dagbag.import_errors) == 0, f"DAG import errors: {dagbag.import_errors}"

def test_dag_structure(dagbag):
    """Test specific DAG structure"""
    dag = dagbag.get_dag('example_etl')

    assert dag is not None
    assert len(dag.tasks) == 3
    assert dag.schedule_interval == '0 6 * * *'

def test_task_dependencies(dagbag):
    """Test task dependencies are correct"""
    dag = dagbag.get_dag('example_etl')

    extract_task = dag.get_task('extract')
    assert 'start' in [t.task_id for t in extract_task.upstream_list]
    assert 'end' in [t.task_id for t in extract_task.downstream_list]

def test_dag_integrity(dagbag):
    """Test DAG has no cycles and is valid"""
    for dag_id, dag in dagbag.dags.items():
        assert dag.test_cycle() is None, f"Cycle detected in {dag_id}"

# Test individual task logic
def test_extract_function():
    """Unit test for extract function"""
    from dags.example_dag import extract_data

    result = extract_data(ds='2024-01-01')
    assert 'records' in result
    assert isinstance(result['records'], int)

测试方法拆解：

• DagBag 解析测试：DagBag(dag_folder='dags/', include_examples=False) 模拟调度器解析目录；import_errors 非空即说明有 DAG 无法被解析（通常是语法错、导入错、模板渲染错），必须在 CI 中拦截。
• 结构与依赖测试：用 dag.get_task('extract') 取出任务，再断言其 upstream_list/downstream_list 符合预期（start → extract → end），防止重构时悄悄改坏拓扑。
• 无环完整性测试：遍历 dagbag.dags 逐个执行 dag.test_cycle()，任何环都会导致调度异常。该测试对应前文 任务依赖 一节中复杂依赖图的正确性保障。
• 纯函数单元测试：把 extract_data 这类可独立执行的函数抽取成普通函数后直接单测（示例中通过 ds='2024-01-01' 显式注入执行日期），业务逻辑不必依赖真实 Airflow 运行时。
• 一个注意点：test_dag_structure 中断言的是旧属性 dag.schedule_interval；在 Airflow 2.4+ 中用 schedule 声明调度后，测试可改断言 dag.schedule，二者取决于你部署的 Airflow 版本。

生产项目目录规范

无论模式如何选，一个可维护的 Airflow 项目应遵循 details.md 给出的目录骨架：

airflow/
├── dags/
│   ├── __init__.py
│   ├── common/
│   │   ├── __init__.py
│   │   ├── operators.py    # Custom operators
│   │   ├── sensors.py      # Custom sensors
│   │   └── callbacks.py    # Alert callbacks
│   ├── etl/
│   │   ├── customers.py
│   │   └── orders.py
│   └── ml/
│       └── training.py
├── plugins/
│   └── custom_plugin.py
├── tests/
│   ├── __init__.py
│   ├── test_dags.py
│   └── test_operators.py
├── docker-compose.yml
└── requirements.txt

组织原则：

• DAG 文件保持轻薄：SKILL.md 明确“不要把重逻辑塞进 DAG 文件”，dags/common/ 下集中放置自定义 Operator、Sensor 与告警回调，供多个 DAG 复用；etl/、ml/ 按业务域划分 DAG。
• tests 与 dags 平行：测试文件按“解析 → 结构 → 无环 → 业务逻辑”分层，可在 CI 中用 pytest 直接跑通。
• plugins/ 目录：存放 Airflow 插件（自定义 Hook、视图、宏等）；custom_plugin.py 中的注册逻辑需与 [plugins_folder] 配置配合。
• docker-compose.yml + requirements.txt：提供本地一键起 Airflow 与依赖锁定，让“本地可测”成为开发默认动作。

Best Practices：Do's 与 Don'ts 清单

SKILL.md 的收尾部分浓缩了实战中血泪经验，这里结合模式示例逐条展开：

应该做（Do's）

• 使用 TaskFlow API：代码更简洁，XCom 自动传递，见模式一。这是 Airflow 2.x 的主推写法。
• 设置超时（timeouts）：为任务设置执行超时与传感器 timeout，防止“僵尸任务”长期占用 worker；长轮询场景务必配 mode='reschedule' 释放槽位。
• Sensor 使用 mode='reschedule'：避免大量等待型 Sensor 阻塞并发池，见模式四。
• 测试 DAG：用 DagBag 做结构/无环/依赖断言，配合纯函数单测，见模式六。
• 保证任务幂等：重试与补跑才能安全，与 DAG 四大原则的 Idempotent 对应。

不要做（Don'ts）

• 不要轻易用 depends_on_past=True：会让某个周期失败后下游周期全部积压成瓶颈；如需串行控制优先用 max_active_runs=1。
• 不要硬编码日期：用 {{ ds }} 宏与执行上下文派生，否则回填、跨时区、历史重跑都会出错。
• 不要使用全局状态：Task 应无状态，共享可变变量在多 worker 部署下行为不可预期（模式二为此专门用工厂隔离配置）。
• 不要盲目跳过 catchup：catchup=False 是常见默认，但要理解其语义——需要补历史数据时需显式 backfill 或按需开启。
• 不要把重逻辑放进 DAG 文件：DAG 文件会被调度器高频重复解析，内部应只做“组 DAG、拼依赖”，业务逻辑放到 dags/common/ 模块或独立包中，见目录规范。

总结与生态用法

airflow-dag-patterns 为 Airflow 生产实践提供了从“四原则 + 依赖语法 + 入门 DAG”到“TaskFlow / 动态 DAG / 分支 / Sensor / 告警 / 测试”的完整闭环。在 agents 仓库中，它通过渐进式披露（SKILL.md → references/details.md）控制上下文成本，并与 data-engineer 智能体（编排与动态 DAG 能力）、data-pipeline.md 命令（管道整体架构与 XCom/重试策略）以及同插件下 spark-optimization、dbt-transformation-patterns、data-quality-frameworks 等技能组合，形成一套“编排 + 计算 + 转换 + 质检”的数据工程工具箱。

阅读本文后，推荐的落地路径是：先用 Quick Start 搭出最小 DAG，按四原则审视幂等与增量，依据数据特征从六个模式中选择合适的组合（如 TaskFlow + Sensor + 分支质检 + 回调告警），最后用 DagBag 测试与无环检查把 DAG 锁进 CI，即可获得一条可重试、可回溯、可观测的生产级调度管道。