Airflow Dag Patterns

作者 wshobson46891e7e60da無授權條款收錄於 2026年10月8日更新於 2026年10月8日

Build production Apache Airflow DAGs with best practices for operators, sensors, testing, and deployment. Use when creating data pipelines, orchestrating workflows, or scheduling batch jobs.

僅含說明DevOps & Cloud Software Development

AI 產生的概覽

指導建置正式環境的 Apache Airflow DAG，涵蓋算子、感測器、測試與部署模式。

功能: 提供面向正式環境的 Apache Airflow DAG 設計指引，涵蓋 DAG 設計原則、任務相依性、算子、感測器、測試與部署策略。內容包含快速入門範例 DAG 以及最佳實務的正反建議，更詳細的模式文件放在參考檔案中。它產出的是說明與程式碼範例，本身不執行任何操作。
適用情境: 適用於建立資料管線協調、設計 DAG 結構與相依性、實作自訂算子或感測器、在本機測試 DAG、在正式環境建置 Airflow，或偵錯失敗的 DAG 執行。
執行需求: 不隨附指令碼，僅為說明文件。參照範例需要 Apache Airflow 環境與 Python，但閱讀這些指引本身不需要任何相依項目。

Apache Airflow DAG Patterns

Production-ready patterns for Apache Airflow including DAG design, operators, sensors, testing, and deployment strategies.

When to Use This Skill

Creating data pipeline orchestration with Airflow
Designing DAG structures and dependencies
Implementing custom operators and sensors
Testing Airflow DAGs locally
Setting up Airflow in production
Debugging failed DAG runs

Core Concepts

1. DAG Design Principles

2. Task Dependencies

python

# Lineartask1 >> task2 >> task3
# Fan-outtask1 >> [task2, task3, task4]
# Fan-in[task1, task2, task3] >> task4
# Complextask1 >> task2 >> task4task1 >> task3 >> task4

# Lineartask1 >> task2 >> task3
# Fan-outtask1 >> [task2, task3, task4]
# Fan-in[task1, task2, task3] >> task4
# Complextask1 >> task2 >> task4task1 >> task3 >> task4

Quick Start

python

# dags/example_dag.pyfrom datetime import datetime, timedeltafrom airflow import DAGfrom airflow.operators.python import PythonOperatorfrom 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

# dags/example_dag.pyfrom datetime import datetime, timedeltafrom airflow import DAGfrom airflow.operators.python import PythonOperatorfrom 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

Detailed patterns and worked examples

Detailed pattern documentation lives in references/details.md. Read that file when the navigation tier above is insufficient.

Best Practices

Do's

Use TaskFlow API - Cleaner code, automatic XCom
Set timeouts - Prevent zombie tasks
Use mode='reschedule' - For sensors, free up workers
Test DAGs - Unit tests and integration tests
Idempotent tasks - Safe to retry

Don'ts

Don't use depends_on_past=True - Creates bottlenecks
Don't hardcode dates - Use {{ ds }} macros
Don't use global state - Tasks should be stateless
Don't skip catchup blindly - Understand implications
Don't put heavy logic in DAG file - Import from modules

來源與署名

來源：wshobson/agents位於plugins/data-engineering/skills/airflow-dag-patterns提交46891e7

授權條款：無授權條款

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架

Principle	Description
Idempotent	Running twice produces same result
Atomic	Tasks succeed or fail completely
Incremental	Process only new/changed data
Observable	Logs, metrics, alerts at every step

Airflow Dag Patterns

Apache Airflow DAG Patterns

When to Use This Skill

Core Concepts

1. DAG Design Principles

2. Task Dependencies

Quick Start

Detailed patterns and worked examples

Best Practices

Do's

Don'ts

來源與署名

更多來自 wshobson/agents 的技能

Uv Package Manager

Web Component Design

Python Type Safety

Visual Design Foundations

Responsive Design

Python Resource Management

更多DevOps & Cloud技能

M5 Onboard

Runbook

Incident Response

Deploy Checklist

Spanner Basics

Secops Cases