テンプレート、冪等性、バックフィル

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日）：

3月31日の売上を INSERT
結果：3行

実行 2（3月31日を再実行）：

3月31日の売上を再度 INSERT
結果：6行（重複発生！）

再実行による重複行

DELETE してから INSERT するパターン

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 による方法も使用できます

DELETE してから INSERT

バックフィル

過去の日付の範囲を再処理します
Airflow は論理日付ごとに1つの DAG 実行を作成します
冪等性と組み合わせることで、バックフィルを安全に実行できます

バックフィルはスケジュールに依存する

UI でのバックフィル

DAG をトリガーし、Backfill を選択して開始日と終了日を設定します
Missing Runs、Missing and Errored Runs、All Runs から実行対象を選択できます
Max Active Runs で並列数を設定し、実行順序を変更することも可能です
Run Backfill でプロセスが開始され、各実行が作成されます

Airflow バックフィル UI

練習しましょう！

Airflow によるデータパイプラインの構築