curiositech/windags-skills

Airflow DAG 編排器

設計並操作 Apache Airflow DAG，以實現具備正確依賴管理、SLA 和監控功能的可靠資料管道編排。

啟用觸發條件

啟用於：「Airflow」、「DAG」、「operator」、「sensor」、「scheduler」、「task dependency」、「SLA」、「backfill」、「XCom」、「TaskFlow API」、「MWAA」、「Cloud Composer」

**不適用於：**dbt 模型執行 → dbt-analytics-engineer（雖然 Airflow 可以觸發 dbt）| 串流處理 → streaming-pipeline-architect | 工作流程引擎（Temporal）→ distributed-transaction-manager

快速入門

1. 定義 DAG — 使用 TaskFlow API（@dag、@task 裝飾器）建立 Python 原生 DAG
2. 設定排程 — cron 或時間表，並設定 catchup=False，除非你有意執行回填
3. 配置重試 — 在每個任務上設定 retries=2、retry_delay=timedelta(minutes=5)
4. 新增 SLA — 在關鍵路徑任務上設定 sla=timedelta(hours=2)
5. 本地測試 — 部署前執行 airflow dags test my_dag 2026-01-01

核心能力

領域	技術
Airflow	Apache Airflow 2.10+, MWAA, Cloud Composer 3
運算子	BashOperator, PythonOperator, KubernetesPodOperator
提供者	apache-airflow-providers-{snowflake, google, aws, dbt-cloud}
執行器	CeleryExecutor, KubernetesExecutor, LocalExecutor
監控	SLA 錯過、任務持續時間、Airflow 指標 → Prometheus

架構模式

TaskFlow API DAG

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule="0 6 * * *",          # 每天早上 6 點 UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(hours=2),
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task()
    def extract_payments() -> dict:
        """從 Stripe API 擷取資料"""
        data = stripe_client.list_payments(date=today())
        return {"count": len(data), "path": "s3://raw/payments/"}

    @task()
    def extract_orders() -> dict:
        """從 Shopify API 擷取資料"""
        data = shopify_client.list_orders(date=today())
        return {"count": len(data), "path": "s3://raw/orders/"}

    @task()
    def transform(payments: dict, orders: dict) -> str:
        """在 DuckDB 中進行連接與轉換"""
        result_path = run_duckdb_transform(payments["path"], orders["path"])
        return result_path

    @task()
    def load(path: str):
        """載入至 Snowflake"""
        snowflake_copy_into("fct_revenue", path)

    # 透過函數呼叫定義依賴
    payments = extract_payments()
    orders = extract_orders()
    transformed = transform(payments, orders)
    load(transformed)

daily_revenue_pipeline()

動態任務映射（扇出/扇入）

@task()
def get_partitions() -> list[str]:
    return ["2026-01-01", "2026-01-02", "2026-01-03"]

@task()
def process_partition(partition_date: str) -> dict:
    """為每個分割區並行執行"""
    return {"date": partition_date, "rows": process(partition_date)}

@task()
def aggregate(results: list[dict]):
    """扇入：接收所有分割區的結果"""
    total = sum(r["rows"] for r in results)
    log.info(f"處理了總共 {total} 行")

# 動態將 process_partition 映射到所有分割區
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)

dbt + Airflow 整合

from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# 選項 1：cosmos（建議使用）
dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/airflow/dbt/"),
    profile_config=ProfileConfig(
        profile_name="default",
        target_name="prod",
    ),
    schedule="@daily",
    dag_id="dbt_daily",
)

# 選項 2：BashOperator（簡單）
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)

反模式

1. 肥大任務 — 任務應為原子單元；不要在同一個任務中執行擷取+轉換+載入
2. 使用 XCom 傳輸大型資料 — XCom 僅用於元資料（路徑、計數）；透過 S3/GCS 傳遞資料，不要使用 XCom pickle
3. 意外啟用 catchup=True — 除非你想執行回填，否則請設定 catchup=False；否則 Airflow 會執行所有錯過的間隔
4. 沒有重試 — 暫時性失敗很常見；務必設定 retries >= 1 並加上延遲
5. DAG 檔案中的頂層程式碼 — DAG 檔案每 30 秒解析一次；模組層級的重型匯入或 API 呼叫會拖慢排程器

品質檢查清單

• 使用 TaskFlow API 建立 Python 原生 DAG（而非傳統的運算子鏈結）
• 除非有意回填，否則設定 catchup=False
• 所有任務已配置重試（至少 1 次重試）
• 關鍵路徑任務已設定 SLA
• XCom 傳遞參考（S3 路徑），而非資料本體
• 使用動態任務映射進行扇出並行處理
• 部署前使用 airflow dags test 進行本地測試
• 任務冪等性：重新執行任務會產生相同結果
• 已配置任務失敗和 SLA 錯過的警報
• 已套用 DAG 標籤以便在 Airflow UI 中過濾