樣板、冪等性與回補

使用 Airflow 建置資料管線

Volker Janz

Senior Developer Advocate at Astronomer

Airflow 中的 Jinja 樣板

@task.bash
def export_data():
    return "cp /data/sales.csv /export/sales_{{ ds }}.csv"

{{ ds }} 會以 YYYY-MM-DD 格式呈現「邏輯日期」
每次執行都有自己的日期，因此同一任務會產生「每次執行不同的輸出」
僅適用於「可套用樣板的欄位」（如 bash_command、sql）

冪等性問題

第 1 次執行（3 月 31 日）：

INSERT 3 月 31 日的 sales
結果：3 列

第 2 次執行（重跑 3 月 31 日）：

再次 INSERT 3 月 31 日的 sales
結果：6 列（重複！）

重跑導致重複列

先刪後插模式

staged_rows = SQLExecuteQueryOperator(
    task_id="get_staged_rows",
    conn_id="duckdb_default",
    sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
)

load = SQLInsertRowsOperator(
    task_id="load_sales",
    conn_id="duckdb_default",
    table_name="sales",
    columns=["date", "product", "amount"],
    preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
    rows=staged_rows.output,
)

也可改用 UPSERT 或 MERGE 做法

先刪後插

回補（Backfilling）

重新處理一段「歷史日期」
Airflow 會為每個邏輯日期建立「一個 Dag 執行」
搭配「冪等性」，回補是「安全的」

回補依賴排程

UI 中的回補

觸發一個 Dag，選擇 Backfill，設定 From 與 To 日期
可選擇觸發 Missing Runs、Missing and Errored Runs，或 All Runs
可用 Max Active Runs 設定平行度，並調整「執行順序」
按下 Run Backfill 後即開始，並建立各次執行

Airflow 回補介面

一起來練習吧！

使用 Airflow 建置資料管線