Ramificare

Introducere în Apache Airflow în Python

Mike Metzger

Data Engineer

Ramificare

Oferă logică condiționată (adică, dacă -> atunci pentru sarcini)
Folosind @task.branch
Execută o funcție Python care returnează ID-ul sarcinii următoare (sau o listă de ID-uri)

Exemplu de ramificare

@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'

Exemplu de ramificare

@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

Vizualizare graf cu ramificare

Vizualizare graf Airflow a unui DAG cu ramificare și două căi de sarcini în aval

Ramificare – luni de sfârșit de trimestru

Vizualizare graf Airflow cu sarcinile de sfârșit de trimestru executate și sarcinile lunare omise

Ramificare – luni obișnuite

Vizualizare graf Airflow cu sarcinile lunare obișnuite executate și sarcinile de sfârșit de trimestru omise

Variabile de dată

ds - Dată logică cu liniuțe YYYY-MM-DD
ds_nodash - Dată logică fără liniuțe YYYYMMDD
prev_data_interval_start_success - Data ultimei rulări DAG reușite
Multe altele – consultați documentația variabilelor Airflow

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

Să exersăm!

Introducere în Apache Airflow în Python