Unknown-333/authoring-airflow-dags

撰寫 Airflow DAG

使用時機

• 建立或重構 Airflow DAG 與任務。
• 設定排程、補跑/回填、重試與 SLA。
• 在任務間傳遞資料（XCom）或使用連線/變數。
• 請勿用於診斷正在執行但中斷的 DAG（請使用 debugging-airflow-pipelines）。

工作流程

- [ ] 讓每個任務具備冪等性，並以資料區間為參數
- [ ] 將耗費資源/大量匯入的程式碼放在任務內部，而非模組頂層
- [ ] 刻意設定排程與補跑
- [ ] 設定重試次數、重試延遲與 SLA
- [ ] 使用 TaskFlow 回傳值或 >> 運算子建立相依關係

1. 冪等任務 — 針對 2026-01-15 區間的任務，無論執行一次或重新執行，都必須產生相同結果。請使用資料區間，而非 datetime.now()。
2. 避免耗時頂層程式碼 — 排程器會頻繁解析每個 DAG 檔案；在模組頂層進行資料庫呼叫、API 呼叫或大型匯入會拖慢排程，甚至導致解析失敗。請將它們放在任務內部。
3. 刻意設定補跑 — catchup=True 會從 start_date 開始回補所有錯過的區間；除非有意為之，否則預設為 False。
4. 重試與 SLA — 暫時性失敗是正常的；設定 retries 與 retry_delay；使用 SLA/警示來處理延遲。

模式

TaskFlow DAG，具冪等性且依賴清晰：

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["orders"],
)
def orders_pipeline():

    @task
    def extract(data_interval_start=None):
        # 使用區間而非 now()，確保重新執行時結果一致。
        return fetch_orders(day=data_interval_start.date())

    @task
    def load(rows):
        # 先刪除再插入分割區 -> 重試時具冪等性。
        overwrite_partition("fct_orders", rows)

    load(extract())

orders_pipeline()

透過 XCom 傳遞小量資料（回傳值）；大量資料則透過儲存服務傳遞 — 寫入 S3/GCS/資料倉儲，並傳遞路徑/金鑰，永遠不要透過 XCom 傳遞數 MB 的資料。

使用連線/變數 來處理機密資訊與設定（BaseHook.get_connection、Variable.get），絕不硬編碼憑證。

常見陷阱

• 頂層 API/資料庫呼叫或大量匯入 — 拖慢排程器，可能導致整個部署環境的 DAG 解析失敗。
• 任務內使用 datetime.now() — 破壞冪等性與補跑功能；請使用 data_interval_start/_end。
• 非預期的 catchup=True — 初次部署時會用歷史執行記錄淹沒叢集。
• 透過 XCom 傳遞大型負載 — 使中繼資料庫膨脹；請傳遞參考。
• 動態 start_date（例如 days_ago）— 導致排程不確定；請使用固定時間戳。
• 單一巨大任務 — 將提取/轉換/載入拆分，以便更精細地進行重試。

參考資料

• 排程、補跑與回填參考