mcpField
A
airflow-dag-patterns

airflow-dag-patterns

github.com/wshobson数据与数据库41.9k11.1k 次安装2 周前更新
代码仓库

关于 airflow-dag-patterns

为使用 Apache Airflow 的数据工程师提供 DAG 设计、测试与部署模式，涵盖任务依赖、运维实践和故障排查。

主要功能

  1. 01DAG 设计原则与依赖模式
  2. 02自定义 Operator 和 Sensor
  3. 03本地 DAG 测试
  4. 04生产部署与故障排查
  5. 05幂等任务与可观测性实践

适用场景

  1. 01构建数据管道编排
  2. 02设计 DAG 结构和任务依赖
  3. 03测试或调试失败的 DAG 运行

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/wshobson/agents --skill airflow-dag-patterns

安装后在 Claude Code 中输入/airflow-dag-patterns确认已加载。

附带文件references/details.md

详细介绍

Apache Airflow DAG 模式

适用于 Apache Airflow 的生产级模式，涵盖 DAG 设计、Operator、Sensor、测试和部署策略。

何时使用本技能

  • 使用 Airflow 创建数据管道编排
  • 设计 DAG 结构和依赖关系
  • 实现自定义 Operator 和 Sensor
  • 在本地测试 Airflow DAG
  • 在生产环境中部署 Airflow
  • 调试失败的 DAG 运行

核心概念

1. DAG 设计原则
原则描述
幂等运行两次产生相同结果
原子任务要么完全成功，要么完全失败
增量仅处理新增/变更的数据
可观测每一步都有日志、指标、告警
2. 任务依赖
# Linear
task1 >> task2 >> task3

# Fan-out
task1 >> [task2, task3, task4]

# Fan-in
[task1, task2, task3] >> task4

# Complex
task1 >> task2 >> task4
task1 >> task3 >> task4

快速开始

# 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

详细模式与完整示例

详细的模式文档位于 references/details.md。当上述导航层级不够用时，请阅读该文件。

最佳实践

推荐做法
  • 使用 TaskFlow API - 代码更简洁，自动使用 XCom
  • 设置超时 - 防止僵尸任务
  • 使用 mode='reschedule' - 用于 Sensor，释放 Worker
  • 测试 DAG - 单元测试和集成测试
  • 幂等任务 - 可安全重试
避免做法
  • 不要使用 depends_on_past=True - 会造成瓶颈
  • 不要硬编码日期 - 使用 {{ ds }} 宏
  • 不要使用全局状态 - 任务应无状态
  • 不要盲目跳过 catchup - 理解其影响
  • 不要把重逻辑放在 DAG 文件中 - 从模块导入

评论

支持 Markdown