分支

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

分支

  • 提供条件逻辑（即 if -> then 用于任务）

  • 使用 @task.branch

  • 运行一个 Python 函数返回下一个要执行的任务 ID（或 ID 列表）

Python 中的 Apache Airflow 入门

分支示例

@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'
Python 中的 Apache Airflow 入门

分支示例

@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
Python 中的 Apache Airflow 入门

分支图视图

Airflow 图视图：具有两个下游路径的分支 DAG

Python 中的 Apache Airflow 入门

分支 季度末月份

Airflow 图视图：季度末任务已运行，月度任务已跳过

Python 中的 Apache Airflow 入门

分支 常规月份

Airflow 图视图：常规月度任务已运行，季度末任务已跳过

Python 中的 Apache Airflow 入门

日期变量

  • ds - 逻辑日期（含连字符）YYYY-MM-DD
  • ds_nodash - 逻辑日期（无连字符）YYYYMMDD
  • prev_data_interval_start_success - 上次成功 DAG 运行的开始日期
  • 还有更多——请查阅 Airflow 变量文档
1 https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#templates-variables
Python 中的 Apache Airflow 入门

让我们来练习！

Python 中的 Apache Airflow 入门

Preparing Video For Download...