用 Jinja 做樣板

Python 中的 Apache Airflow 入門

Mike Metzger

Data Engineer

什麼是樣板？

  • 一種定義好的格式，讓 Airflow 自動代入資訊
  • 常為純文字／文字檔
  • 簡化存取與重複使用

 

含可代入值佔位符的樣板示意圖

Python 中的 Apache Airflow 入門

什麼是 Jinja？

  • 簡單的文字型樣板語言
  • 不只 Airflow，許多工具都在用
  • {{ ... }} 表示樣板代入
    • 類似 Python 的 f-string
  • 可更動態，支援 for 迴圈

使用雙大括號代入的 Jinja 樣板示意圖

Python 中的 Apache Airflow 入門

Airflow 的 Jinja 函式

  • logical_date：Dag 執行發生時間
  • ds：logical_date 的 YYYY-MM-DD 格式
  • logical_date.year、logical_date.month、logical_date.day：存取日期組件
  • params：Dag 執行的執行階段參數
  • 可做計算、迴圈等
    • macros.ds_add：進行日期運算
Python 中的 Apache Airflow 入門

使用 SmtpNotifier 的範例

  • 在主旨欄位中代入
@dag(dag_id=`sales_etl`,
     on_failure_callback=SmtpNotifier(
         to='[email protected]',
         from_email='[email protected]',

         subject='sales_etl on {{ logical_date }} has failed!'
     )
)
  • Subject: sales_etl on 2026-04-30 failed!
Python 中的 Apache Airflow 入門

使用 FileSensor 的範例

  • 動態設定檔案路徑
  • 尋找 ds 前一天（logical_date，格式為 YYYY-MM-DD）的檔案
  • 使用 macros.ds_add
    filewatcher = FileSensor(
task_id="wait_for_files",
filepath="/data/{{ macros.ds_add(ds, -1) }}/input.csv",
...
)
  • 若 ds 為 2026-04-30，檔案路徑為 /data/2026-04-29/input.csv
Python 中的 Apache Airflow 入門

Jinja 注意事項

  • 不是到處都能用
  • 請查看文件中的「可樣板化」屬性
  • 檢視程式碼檢視畫面

Airflow Dag 程式碼檢視，顯示定義 Dag 的唯讀 Python 原始碼

Python 中的 Apache Airflow 入門

一起來練習吧！

Python 中的 Apache Airflow 入門

Preparing Video For Download...