Dags robustos com tratamento de falhas e novas tentativas

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Por que tarefas falham

Problemas de conexão de tarefas

Timeouts e limites de taxa de API
Interrupções rápidas de rede e atrasos de DNS
Várias tarefas competindo pelo mesmo recurso

Novas tentativas (retries)

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
)
def fetch_weather():
    response = requests.get("https://api.weather.com/forecast")
    response.raise_for_status()
    return response.json()

A tarefa tenta novamente até 3 vezes em caso de falha
Espera 2 minutos entre as tentativas

Backoff exponencial

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
    retry_exponential_backoff=2.0,
)
def fetch_weather():
    ...

Retrocesso exponencial

O atraso dobra após cada nova tentativa
Dá tempo para serviços externos se recuperarem

on_failure_callback

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")

@task(on_failure_callback=alert_on_failure)
def fetch_weather():
    ...

Dict context: dag, ti, exception, log URL
Direcione para Slack, PagerDuty ou outra ferramenta de alerta

Callbacks nos níveis Dag vs tarefa

Nível de Dag (abrangente)

@dag(
    on_failure_callback=alert_team,
)
def my_pipeline():
    ...

Nível de tarefa (específico)

@task(
    on_failure_callback=page_oncall,
)
def critical_step():
    ...

No nível de tarefa sobrepõe o de Dag
Use ambos para alertas em camadas

max_consecutive_failed_dag_runs

@dag(max_consecutive_failed_dag_runs=3)
def monitoring_pipeline():
    ...

Execuções de Dag com falha seguidas de pausa automática

Evita a fadiga de alertas por falhas persistentes
O Dag volta quando você despausar manualmente

Juntando tudo

@dag(
    max_consecutive_failed_dag_runs=3,
    on_failure_callback=alert_team,
)
def weather_pipeline():

    @task(
        retries=3,
        retry_delay=timedelta(minutes=2),
        retry_exponential_backoff=True,
        on_failure_callback=page_oncall,
    )
    def fetch_weather():
        ...

Retries + backoff tratam falhas transitórias
Callbacks alertam as pessoas certas
Pausa automática reduz o barulho

Vamos praticar!

Criando Pipelines de Dados com Airflow