Acionadores e tolerância a falhas

Introdução ao Apache Airflow em Python

Mike Metzger

Data Engineer

Criando pipelines robustos

Triggers - controlam quando as tarefas podem rodar
- Definem tarefas que executam sob condições específicas
- Trigger padrão - tudo com sucesso

Ilustração de triggers controlando quando tarefas podem rodar

Tolerância a falhas - permite recuperação dentro de uma tarefa
- Tenta novamente a tarefa individual

Ilustração de tolerância a falhas repetindo uma tarefa com erro

Regras de acionamento

Verifica o estado das tarefas anteriores antes de iniciar
Permite definir se as tarefas devem prosseguir
Por padrão, todas as anteriores devem terminar com sucesso para continuar
Regras de acionamento permitem mudar isso

Ilustração de trigger rules verificando estados upstream

Usos de trigger rules

Tarefas de notificação
Tarefas de limpeza
Execução condicional

Ilustração de usos de trigger rules como notificação e limpeza

Principais trigger rules

all_success - Todas as tarefas anteriores concluíram com sucesso, sem skips
all_failed - Todas as tarefas anteriores falharam
all_done - Todas as tarefas anteriores concluídas, qualquer que seja o resultado
one_failed - Pelo menos uma falhou
one_success - Pelo menos uma teve sucesso
none_failed - Todas as anteriores concluíram com sucesso ou foram puladas

Implementação de trigger rule

from airflow.utils.trigger_rule import TriggerRule
Regras de acionamento são aplicadas ao decorador @task
trigger_rule=TriggerRule.<TriggerRule enum>

@task(trigger_rule=TriggerRule.ALL_SUCCESS)
def run_if_everything_succeeds:
  print('All previous tasks succeeded!')

Atributos de tolerância a falhas

retries - parâmetro que define quantas vezes o Airflow deve tentar novamente antes de marcar como falha
retry_delay - aceita um timedelta com o tempo de espera entre tentativas

@task(
  trigger_rule=TriggerRule.ONE_SUCCESS,
  retries=3,
  retry_delay=timedelta(minutes=5)
)

TriggerDagRunOperator

Permite que um DAG dispare outro DAG
Disponibiliza código e DAGs
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
trigger_dag_id - Deve corresponder ao dag_id
wait_for_completion - Bloqueia tarefas seguintes até concluir
poke_interval - Com que frequência verificar
conf - Dados passados ao DAG filho

Exemplo de TriggerDagRun

trigger_child = TriggerDagRunOperator(
        task_id="trigger_child_pipeline",
        trigger_dag_id="child_pipeline_dag",   
        wait_for_completion=True,              
        poke_interval=30,                      
        conf={                                 
            "source": "s3://my-bucket/raw/",
        },
    )

task1 >> trigger_child    # Executa a primeira tarefa e então dispara o DAG filho
trigger_child >> cleanup  # Executa limpeza após o DAG filho concluir

Vamos praticar!

Introdução ao Apache Airflow em Python