Розгалуження

Вступ до Apache Airflow на Python

Mike Metzger

Data Engineer

Розгалуження

Забезпечує умовну логіку (тобто if -> then для задач)
Використання @task.branch
Запускає функцію Python, яка повертає ідентифікатор наступної задачі (або список 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

Графічний вигляд розгалуження

Графічний вигляд гілкування DAG в Airflow із двома нижчими шляхами задач

Розгалуження: місяці кінця кварталу

Графік 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

Давайте потренуємось!

Вступ до Apache Airflow на Python