Ramificação

Introdução ao Apache Airflow em Python

Mike Metzger

Data Engineer

Ramificação

Ramificação no Airflow:

Fornece lógica condicional
Usando BranchPythonOperator
from airflow.operators.python import BranchPythonOperator
Recebe um python_callable que retorna o id da próxima tarefa (ou lista de ids) a seguir

Exemplo de ramificação

def branch_test(**kwargs):
  if int(kwargs['ds_nodash']) % 2 == 0:
    return 'even_day_task'
  else:
    return 'odd_day_task'

Exemplo de ramificação

def branch_test(**kwargs):
  if int(kwargs['ds_nodash']) % 2 == 0:
    return 'even_day_task'
  else:
    return 'odd_day_task'


branch_task = BranchPythonOperator(task_id='branch_task',dag=dag,
       provide_context=True,
       python_callable=branch_test)


start_task >> branch_task >> even_day_task >> even_day_task2

branch_task >> odd_day_task >> odd_day_task2

Visão do grafo de ramificação

Visão do grafo do DAG com ramificação

Ramificação: dias pares

Visão do grafo de ramificação - dias pares

Ramificação: dias ímpares

Visão do grafo de ramificação - dias ímpares

Vamos praticar!

Introdução ao Apache Airflow em Python