分支

Python 中的 Apache Airflow 入門

Mike Metzger

Data Engineer

分支

提供條件邏輯（例如 if -> then 的任務流程）
使用 @task.branch
執行 Python 函式，回傳下一個要執行的任務 ID（或 ID 清單）

分支範例

@task.branch
def branch_task(logical_date):
  if int(logical_date.month) % 3 == 0:
    # Months 3 - March, 6 - June, 9 - September, 12 - December
    return 'end_of_quarter_task'
  else:
    # All other months
    return 'regular_monthly_task'

分支範例

@task.branch
def branch_task(logical_date):
  if int(logical_date.month) % 3 == 0:
    # Months 3 - March, 6 - June, 9 - September, 12 - December
    return 'end_of_quarter_task'
  else:
    # All other months
    return 'regular_monthly_task'


start_task >> branch_task >> end_of_quarter_task >> end_of_quarter_task2

branch_task >> regular_monthly_task >> regular_monthly_task2

分支圖形檢視

Airflow 分支 Dag 的圖形檢視，含兩條下游任務路徑

分支：季末月份

Airflow 圖形檢視：季末任務已執行，月任務被略過

分支：一般月份

Airflow 圖形檢視：一般月任務已執行，季末任務被略過

日期變數

ds－帶連字號的邏輯日期 YYYY-MM-DD
ds_nodash－不帶連字號的邏輯日期 YYYYMMDD
prev_data_interval_start_success－上一次成功 Dag 執行的日期
還有更多－請查閱 Airflow 變數文件

¹ https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#templates-variables

一起來練習吧！

Python 中的 Apache Airflow 入門