Como solucionar problemas do programador do Airflow

Airflow gerenciado (Geração 3) | Airflow gerenciado (Geração 2) | Airflow gerenciado (Geração 1 legada)

Nesta página, você encontra etapas de solução de problemas e informações sobre problemas comuns com programadores do Airflow e processadores de DAG.

Identificar a origem do problema

Para começar a solução de problemas, identifique se o problema ocorre:

  • No momento da análise do DAG, enquanto o DAG é analisado por um processador de DAG do Airflow
  • No momento da execução, enquanto o DAG é processado por um programador do Airflow

Para mais informações sobre o tempo de análise e o tempo de execução do DAG, leia Diferença entre o tempo de análise do DAG e o tempo de execução do DAG.

Inspecionar problemas de processamento de DAG

  1. Inspecione os registros do processador de DAG.
  2. Verifique os tempos de análise do DAG.

Como monitorar tarefas em execução ou na fila

Para verificar se há tarefas travadas em uma fila, siga estas etapas.

  1. No Google Cloud console do Cloud, acesse a página Ambientes.

    Acessar "Ambientes"

  2. Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.

  3. Acesse a guia Monitoramento.

  4. Na guia Monitoramento, consulte o gráfico Tarefas do Airflow na seção Execuções do DAG e identifique possíveis problemas. As tarefas do Airflow são tarefas que estão em um estado enfileirado no Airflow. Elas podem ir para a fila de agentes do executor do Celery ou do Kubernetes. As tarefas em fila do Celery são instâncias de tarefas colocadas na fila de agentes do Celery.

Solução de problemas no momento da análise do DAG

As seções a seguir descrevem sintomas e possíveis correções para alguns problemas comuns no tempo de análise do DAG.

Número e distribuição de tempo das tarefas

O Airflow pode ter problemas ao programar um grande número de DAGs ou tarefas ao mesmo tempo. Para evitar problemas com a programação, você pode:

  • Ajuste seus DAGs para usar um número menor de tarefas mais consolidadas.
  • Ajuste os intervalos de programação dos DAGs para distribuir as execuções de DAG de maneira mais uniforme ao longo do tempo.

Como escalonar a configuração do Airflow

O Airflow oferece opções de configuração que controlam quantas tarefas e DAGs o Airflow pode executar ao mesmo tempo. Para definir essas opções de configuração, modifique os valores para o ambiente. Também é possível definir alguns desses valores no nível do DAG ou da tarefa.

  • Simultaneidade do worker

    O parâmetro [celery]worker_concurrency controla o número máximo de tarefas que um worker do Airflow pode executar ao mesmo tempo. Se você multiplicar o valor desse parâmetro pelo número de workers do Airflow no ambiente do Airflow gerenciado, você receberá o número máximo de tarefas que podem ser executadas em um determinado momento no ambiente. Esse número é limitado pela opção de configuração do Airflow [core]parallelism, que é descrita em mais detalhes.

    Em ambientes do Airflow gerenciado (Geração 3), o valor padrão de [celery]worker_concurrency é calculado automaticamente com base no número de instâncias de tarefas simultâneas leves que um worker pode acomodar. Isso significa que o valor depende dos limites de recursos do worker. O valor de simultaneidade do worker não depende do número de workers no ambiente.

  • Máximo de execuções de DAGs ativas

    A opção de configuração do Airflow [core]max_active_runs_per_dag controla o número máximo de execuções ativas de DAGs por DAG. O programador não criará mais execuções de DAGs se atingir esse limite.

    Se esse parâmetro for definido incorretamente, você poderá encontrar um problema em que o programador restringe a execução do DAG, porque não é possível criar mais instâncias de execução do DAG em um determinado momento.

    Também é possível definir esse valor no nível do DAG com o parâmetro max_active_runs.

  • Número máximo de tarefas ativas por DAG

    A opção de configuração do Airflow [core]max_active_tasks_per_dag controla o número máximo de instâncias de tarefa que podem ser executadas simultaneamente em cada DAG.

    Se esse parâmetro for definido incorretamente, você poderá encontrar um problema em que a execução de uma única instância do DAG é lenta porque há apenas um número limitado de tarefas do DAG que podem ser executadas em um determinado momento Nesse caso, é possível aumentar o valor dessa opção de configuração.

    Também é possível definir esse valor no nível do DAG com o parâmetro max_active_tasks.

    É possível usar max_active_tis_per_dag e max_active_tis_per_dagrun parâmetros no nível da tarefa para controlar quantas instâncias com um ID de tarefa específico podem ser executadas por DAG e por execução de DAG.

  • Paralelismo e tamanho do pool

    A opção de configuração [core]parallelism do Airflow controla quantas tarefas o programador do Airflow pode enfileirar na fila do executor após todas as dependências dessas tarefas serem atendidas.

    Este é um parâmetro global para toda a configuração do Airflow.

    As tarefas são enfileiradas e executadas em um pool. Os ambientes do Airflow gerenciado usam apenas um pool. O tamanho desse pool controla quantas tarefas podem ser enfileiradas pelo programador para execução em um determinado momento. Se o tamanho do pool for muito pequeno, o programador não poderá enfileirar tarefas para execução, mesmo que os limites sejam definidos pela opção de configuração [core]parallelism e pelo . opção de configuração [celery]worker_concurrency multiplicada pelo número de workers do Airflow ainda não foi atendida.

    É possível configurar o tamanho do pool na interface do Airflow (Admin > Pools). Ajuste o tamanho do pool com o nível de paralelismo esperado no ambiente.

    Normalmente, [core]parallelism é definido como um produto do número máximo de workers e [celery]worker_concurrency.

Solução de problemas com tarefas em execução e na fila

As seções a seguir descrevem sintomas e possíveis correções para alguns problemas comuns com tarefas em execução e na fila.

As execuções de DAG não são executadas

Sintoma:

Quando uma data de programação para um DAG é definida dinamicamente, isso pode levar a vários efeitos colaterais inesperados. Exemplo:

  • Uma execução de DAG está sempre no futuro, e o DAG nunca é executado.

  • As execuções de DAG anteriores são marcadas como executadas e bem-sucedidas, mesmo que não tenham sido executadas.

Mais informações estão disponíveis na documentação do Apache Airflow.

Soluções possíveis:

  • Siga as recomendações na documentação do Apache Airflow.

  • Defina start_date estático para DAGs. Como opção, é possível usar catchup=False para desativar a execução do DAG para datas anteriores.

  • Evite usar datetime.now() ou days_ago(<number of days>), a menos que você esteja ciente dos efeitos colaterais dessa abordagem.

Como usar o recurso de tabela de horários do programador do Airflow

As tabelas de horários estão disponíveis a partir do Airflow 2.2.

É possível definir uma tabela de horários para um DAG com um dos seguintes métodos:

Também é possível usar tabelas de horários integradas.

Evite programar tarefas durante janelas de manutenção

É possível definir janelas de manutenção para o ambiente para que a manutenção do ambiente ocorra fora dos horários em que você executa os DAGs. Ainda é possível executar os DAGs durante as janelas de manutenção, desde que seja aceitável que algumas tarefas possam ser interrompidas e repetidas. Para mais informações sobre como as janelas de manutenção afetam o ambiente, consulte Especificar janelas de manutenção.

Uso de "wait_for_downstream" nos DAGs

Se você definir o parâmetro wait_for_downstream como True nos DAGs, para que uma tarefa seja bem-sucedida, todas as tarefas que estiverem imediatamente downstream também serão bem-sucedidas. Isso significa que a execução de tarefas pertencentes a uma determinada execução do DAG pode ser reduzida pela execução de tarefas da execução anterior do DAG. Leia mais sobre isso em a documentação do Airflow.

As tarefas enfileiradas por muito tempo serão canceladas e reprogramadas

Se uma tarefa do Airflow for mantida na fila por muito tempo, o programador a reprogramará para execução após o período definido na opção de configuração do Airflow [scheduler]task_queued_timeout. O valor padrão é 2400. Em versões do Airflow anteriores à 2.3.1, a tarefa também é marcada como com falha e repetida se for qualificada para uma nova tentativa.

Uma maneira de observar os sintomas dessa situação é analisar o gráfico com o número de tarefas enfileiradas (guia "Monitoramento" na interface do Airflow gerenciado). Se os picos nesse gráfico não caírem em cerca de duas horas, as tarefas provavelmente serão reprogramadas (sem registros), seguidas pelas entradas de registro "As tarefas adotadas ainda estavam pendentes..." nos registros do programador. Nesses casos, a mensagem "O arquivo de registro não foi encontrado..." pode aparecer nos registros de tarefas do Airflow porque a tarefa não foi executada.

Em geral, esse comportamento é esperado, e a próxima instância da tarefa programada deve ser executada de acordo com a programação. Se você observar muitos casos desse tipo nos ambientes do Airflow gerenciado, isso pode significar que não há workers do Airflow suficientes no ambiente para processar todas as tarefas programadas.

Resolução: para resolver esse problema, verifique se sempre há capacidade nos workers do Airflow para executar tarefas enfileiradas. Por exemplo, é possível aumentar o número de workers ou a simultaneidade do worker. Também é possível ajustar o paralelismo ou os pools para evitar o enfileiramento de tarefas além da capacidade.

As tarefas travadas na fila podem bloquear a execução de um DAG específico

Para resolver esse problema, faça upgrade do ambiente para a versão 2.1.12 ou mais recente do Airflow gerenciado.

Em casos normais, o programador do Airflow precisa lidar com situações em que há tarefas na fila e, por algum motivo, não é possível executá-las corretamente (por exemplo, quando um DAG a que essas tarefas pertencem foi excluído).

Se essas tarefas não forem limpas pelo programador, talvez seja necessário excluí-las manualmente. É possível fazer isso, por exemplo, na interface do Airflow (Menu > Navegador > Instâncias de tarefas), encontrar tarefas enfileiradas e excluí-las.

Abordagem do Airflow gerenciado para o parâmetro min_file_process_interval

O Airflow gerenciado muda a maneira como [scheduler]min_file_process_interval é usado pelo programador do Airflow.

Em versões do Airflow gerenciado anteriores à 2.0.26, [scheduler]min_file_process_interval é ignorado.

Em versões do Airflow gerenciado posteriores à 2.0.26:

O programador do Airflow é reiniciado depois que todos os DAGs são programados um determinado número de vezes, e o [scheduler]num_runs parâmetro controla quantas vezes isso é feito pelo programador. Quando o programador atinge os loops de programação [scheduler]num_runs, ele é reiniciado. O programador é um componente sem estado, e essa reinicialização é um mecanismo de recuperação automática para qualquer problema que o programador possa ter. O valor padrão de [scheduler]num_runs é 5000.

[scheduler]min_file_process_interval pode ser usado para configurar a frequência com que a análise do DAG ocorre, mas esse parâmetro não pode ser maior que o tempo necessário para que um programador execute loops [scheduler]num_runs ao programar os DAGs.

Como marcar tarefas como com falha após atingir dagrun_timeout

O programador marca as tarefas que não estão concluídas (em execução, programadas e enfileiradas) como com falha se uma execução de DAG não terminar dentro de dagrun_timeout (um parâmetro de DAG).

Solução:

Sintomas do banco de dados do Airflow sob carga pesada

Às vezes, nos registros do programador do Airflow, você pode encontrar a seguinte entrada de registro de aviso:

Scheduler heartbeat got an exception: (_mysql_exceptions.OperationalError) (2006, "Lost connection to MySQL server at 'reading initial communication packet', system error: 0")"

Sintomas semelhantes também podem ser observados nos registros de worker do Airflow:

Para MySQL:

(_mysql_exceptions.OperationalError) (2006, "Lost connection to MySQL server at
'reading initial communication packet', system error: 0")"

Para PostgreSQL:

psycopg2.OperationalError: connection to server at ... failed

Esses erros ou avisos podem ser um sintoma de que o banco de dados do Airflow está sobrecarregado pelo número de conexões abertas ou pelo número de consultas executadas ao mesmo tempo, seja por programadores ou por outros componentes do Airflow, como workers, engatilhadores e servidores da Web.

Soluções possíveis:

O servidor da Web mostra o aviso "O programador não parece estar em execução"

O programador informa o sinal de funcionamento regularmente ao banco de dados do Airflow. Com base nessas informações, o servidor da Web do Airflow determina se o programador está ativo.

Às vezes, se o programador estiver sob carga pesada, ele poderá não conseguir informar o sinal de funcionamento a cada [scheduler]scheduler_heartbeat_sec.

Nessa situação, o servidor da Web do Airflow poderá mostrar o seguinte aviso:

The scheduler does not appear to be running. Last heartbeat was received <X>
seconds ago.

Soluções possíveis:

  • Aumente os recursos de CPU e memória para o programador.

  • Otimize os DAGs para que a análise e a programação sejam mais rápidas e não consumam muitos recursos do programador.

  • Evite usar variáveis globais em DAGs do Airflow. Em vez disso, use variáveis de ambiente e variáveis do Airflow.

  • Aumente o valor da opção de configuração do Airflow [scheduler]scheduler_health_check_threshold para que o servidor da Web espere mais tempo antes de informar a indisponibilidade do programador.

Soluções alternativas para problemas encontrados durante o preenchimento de DAGs

Às vezes, é possível executar novamente DAGs que já foram executados. É possível fazer isso com um comando da CLI do Airflow da seguinte maneira:

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location LOCATION \
   dags backfill -- -B \
   -s START_DATE \
   -e END_DATE \
   DAG_NAME

Para executar novamente apenas as tarefas com falha de um DAG específico, use também o argumento --rerun-failed-tasks.

Substitua:

  • ENVIRONMENT_NAME pelo nome do ambiente
  • LOCATION pela região em que o ambiente está localizado
  • START_DATE por um valor para o parâmetro start_date do DAG, no formato YYYY-MM-DD
  • END_DATE por um valor para o parâmetro end_date do DAG, no formato YYYY-MM-DD
  • DAG_NAME pelo nome do DAG

A operação de preenchimento pode gerar uma situação de deadlock em que um preenchimento não é possível porque há um bloqueio em uma tarefa. Exemplo:

2022-11-08 21:24:18.198 CET DAG ID Task ID Run ID Try number
2022-11-08 21:24:18.201 CET -------- --------- -------- ------------
2022-11-08 21:24:18.202 CET 2022-11-08 21:24:18.203 CET These tasks are deadlocked:
2022-11-08 21:24:18.203 CET DAG ID Task ID Run ID Try number
2022-11-08 21:24:18.204 CET ----------------------- ----------- ----------------------------------- ------------
2022-11-08 21:24:18.204 CET <DAG name> <Task name> backfill__2022-10-27T00:00:00+00:00 1
2022-11-08 21:24:19.249 CET Command exited with return code 1
...
2022-11-08 21:24:19.348 CET Failed to execute job 627927 for task backfill

Em alguns casos, é possível usar as seguintes soluções alternativas para superar deadlocks:

A seguir