Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Neste guia, mostramos como criar um gráfico acíclico dirigido (DAG, na sigla em inglês) do Apache Airflow que é executado em um ambiente do Cloud Composer.
Como o Apache Airflow não fornece um forte isolamento de tarefas e DAGs, recomendamos que você use ambientes de produção e teste separados para evitar a interferência de DAGs. Para mais informações, consulte Como testar DAGs.
Como estruturar um DAG do Airflow
Um DAG do Airflow é definido em um arquivo Python e composto pelos seguintes componentes:
- Definição do DAG
- Operadores do Airflow
- Relacionamentos entre operadores
Os snippets de código abaixo mostram exemplos de cada componente fora de contexto.
Uma definição de DAG
O exemplo a seguir demonstra uma definição do Airflow DAG:
Operadores e tarefas
Os operadores do Airflow descrevem o trabalho a ser feito. Uma tarefa é uma instância específica de um operador.
Relações entre tarefas
As relações de tarefas descrevem a ordem em que o trabalho precisa ser concluído.
Exemplo completo de fluxo de trabalho do DAG em Python
O fluxo de trabalho a seguir é um modelo de DAG completo e funcional que é composto por duas tarefas: uma tarefa hello_python
e uma goodbye_bash
:
Para mais informações sobre como definir DAGs do Airflow, consulte o tutorial e os conceitos do Airflow.
Operadores do Airflow
Nos exemplos a seguir, você vê alguns operadores conhecidos do Airflow. Para uma referência autoritativa dos operadores do Airflow, consulte a Referência de operadores e hooks e o Índice de provedores.
BashOperator
Use o BashOperator para executar programas de linha de comando.
O Cloud Composer executa os comandos fornecidos em um script Bash em um worker do Airflow. O worker é um contêiner do Docker baseado em Debian e inclui vários pacotes.
PythonOperator
Use o PythonOperator para executar o código arbitrário do Python.
O Cloud Composer executa o código Python em um contêiner que inclui pacotes para a versão de imagem do Cloud Composer usada no ambiente.
Para instalar mais pacotes Python, consulte Como instalar dependências do Python.
Operadores do Google Cloud
Para executar tarefas que usam produtos do Google Cloud, use os operadores do Google Cloud Airflow. Por exemplo, os operadores do BigQuery consultam e processam dados no BigQuery.
Há muitos outros operadores do Airflow para o Google Cloud e serviços individuais fornecidos pelo Google Cloud. Consulte Operadores do Google Cloud para ver a lista completa.
EmailOperator
Use o EmailOperator para enviar e-mails de um DAG. Para enviar e-mails de um ambiente do Cloud Composer, configure o ambiente para usar o SendGrid.
Notificações sobre falha do operador
Defina email_on_failure
como True
para enviar uma notificação por e-mail quando um operador no DAG falhar. Para enviar notificações por e-mail de um ambiente do Cloud Composer, você precisa configurar o ambiente para usar o SendGrid.
Diretrizes de fluxo de trabalho do DAG
Coloque as bibliotecas Python personalizadas em um arquivo ZIP do DAG em um diretório aninhado. Não inclua bibliotecas no nível superior do diretório de DAGs.
Quando o Airflow verifica a pasta
dags/
, ele verifica apenas os DAGs nos módulos Python que estão no nível superior da pasta de DAGs e de um arquivo ZIP também localizado na pastadags/
de nível superior. Se o Airflow encontrar um módulo do Python em um arquivo ZIP que não contenha as substringsairflow
eDAG
, ele interromperá o processamento do arquivo ZIP. O Airflow retorna apenas os DAGs encontrados até esse ponto.Para ter tolerância a falhas, não defina vários objetos de DAG no mesmo módulo do Python.
Não use SubDAGs. Em vez disso, agrupe tarefas em DAGs.
Coloque os arquivos necessários no tempo de análise do DAG na pasta
dags/
, não na pastadata/
.Teste os DAGs desenvolvidos ou modificados conforme recomendado nas instruções de teste de DAGs.
A ferramenta CLI de desenvolvimento local do Composer simplifica o desenvolvimento de DAGs do Apache Airflow para o Cloud Composer 2 executando um ambiente do Airflow localmente. Esse ambiente local do Airflow usa uma imagem de uma versão específica do Cloud Composer 2.
Verificar se os DAGs desenvolvidos não aumentam muito os tempos de análise dos DAGs.
As tarefas do Airflow podem falhar por vários motivos. Para evitar falhas em execuções inteiras do DAG, recomendamos ativar novas tentativas da tarefa. Definir o máximo de tentativas como
0
significa que nenhuma nova tentativa será realizada.Recomendamos substituir a opção
default_task_retries
por um valor para as tentativas de tarefa diferentes de0
. Além disso, você pode definir o parâmetroretries
no nível da tarefa.Se você quiser usar a GPU nas tarefas do Airflow, crie um cluster do GKE separado com base nos nós que usam máquinas com GPUs. Use o GKEStartPodOperator para executar as tarefas.
Evite executar tarefas com uso intenso de CPU e memória no pool de nós do cluster em que outros componentes do Airflow (programadores, workers e servidores da Web) estão sendo executados. Em vez disso, use KubernetesPodOperator ou GKEStartPodOperator.
Ao implantar DAGs em um ambiente, faça upload apenas dos arquivos que são absolutamente necessários para interpretar e executar DAGs na pasta
/dags
.Limite o número de arquivos DAG na pasta
/dags
.O Airflow está analisando continuamente os DAGs na pasta
/dags
. A análise é um processo que percorre a pasta de DAGs, e o número de arquivos que precisam ser carregados (com as respectivas dependências) afeta o desempenho da análise do DAG e da programação de tarefas. É muito mais eficiente usar 100 arquivos com 100 DAGs cada do que 10.000 arquivos com 1 DAG. Portanto, essa otimização é recomendada. Essa otimização é um equilíbrio entre analisar o tempo e a eficiência da criação e do gerenciamento de DAGs.Por exemplo, para implantar 10.000 arquivos DAG, crie 100 arquivos ZIP, cada um contendo 100 arquivos DAG.
Além das dicas acima, se você tem mais de 10.000 arquivos DAG, a geração de DAGs de maneira programática pode ser uma boa opção. Por exemplo, é possível implementar um único arquivo DAG do Python que gera um certo número de objetos DAG (por exemplo, 20.100 objetos DAG).
Perguntas frequentes sobre como escrever DAGs
Como diminuir a repetição de código para executar tarefas iguais ou semelhantes em vários DAGs?
Sugerimos definir bibliotecas e wrappers para minimizar a repetição de código.
Como faço para reutilizar código entre arquivos de DAGs?
Coloque as funções utilitárias em uma
biblioteca local do Python
e as importe. É possível referenciar as funções em qualquer DAG localizado na pasta dags/
no bucket do ambiente.
Como diminuo o risco de surgirem diferentes definições?
Por exemplo, tenho duas equipes que querem agregar dados brutos em métricas de receita. Elas escrevem duas tarefas ligeiramente diferentes que fazem a mesma coisa. Defina bibliotecas para trabalhar com os dados de receita. Assim, os implementadores de DAGs precisam esclarecer a definição da receita que está sendo agregada.
Como faço para definir dependências entre DAGs?
Isso depende de como você quer definir a dependência.
Se você tiver dois DAGs (DAG A e DAG B) e quiser que o DAG B seja acionado após o DAG A, coloque um TriggerDagRunOperator
no final do DAG A.
Se o DAG B depender somente de um artefato gerado pelo DAG A, como uma mensagem do Pub/Sub, um sensor funcionará melhor.
Se o DAG B estiver muito integrado ao DAG A, talvez seja possível mesclar os dois em um único DAG.
Como transmito códigos exclusivos de execução para um DAG e às tarefas dele?
Por exemplo, quero transmitir caminhos de arquivo e nomes de cluster do Dataproc.
É possível retornar str(uuid.uuid4())
em PythonOperator
para gerar um ID exclusivo aleatório. Isso coloca o ID em
XComs
para que você possa consultá-lo em outros operadores
por meio de campos com modelo.
Antes de gerar um uuid
, considere se um ID específico do DagRun seria mais importante. Também é possível referenciar esses IDs nas substituições por meio de macros.
Como separar tarefas em um DAG?
Cada tarefa precisa ser uma unidade de trabalho idempotente. Consequentemente, evite encapsular um fluxo de trabalho de várias etapas em uma única tarefa, como um programa complexo em execução em um PythonOperator
.
É recomendado definir várias tarefas em um único DAG para agregar dados de várias origens?
Por exemplo, tenho várias tabelas com dados brutos e quero criar agregações diárias para cada uma delas. As tarefas não são dependentes entre si. Preciso criar uma tarefa e um DAG para cada tabela ou gerar um DAG geral?
Se você está satisfeito com cada tarefa compartilhando as mesmas propriedades no nível do DAG, como schedule_interval
, faz sentido definir várias tarefas em um único DAG. Caso contrário, para minimizar a repetição de código, vários DAGs podem ser gerados a partir de um único módulo Python. Basta colocá-los no globals()
do módulo.
Como limitar o número de tarefas simultâneas em execução em um DAG?
Por exemplo, quero evitar exceder os limites/cotas de uso da API ou impedir a execução de muitos processos simultâneos.
É possível definir pools do Airflow na UI da Web do Airflow e associar tarefas aos pools atuais nos DAGs.
Perguntas frequentes sobre o uso de operadores
Devo usar DockerOperator
?
Não recomendamos o uso
do DockerOperator
, a menos que ele seja usado para iniciar
contêineres em uma instalação remota do Docker (não no cluster de um
ambiente). Em um ambiente do Cloud Composer, o operador não tem acesso aos daemons do Docker.
Em vez disso, use KubernetesPodOperator
ou
GKEStartPodOperator
. Esses operadores iniciam os pods do Kubernetes
nos clusters do Kubernetes ou do GKE, respectivamente. Não recomendamos iniciar pods no cluster de um ambiente, porque isso pode levar à concorrência de recursos.
Devo usar SubDagOperator
?
Não recomendamos o uso de SubDagOperator
.
Use alternativas conforme sugerido em Como agrupar tarefas.
Posso executar o código do Python somente em PythonOperators
para separar completamente os operadores Python?
Dependendo do seu objetivo, você tem algumas opções.
Se a única preocupação for manter dependências separadas do Python, use PythonVirtualenvOperator
.
Considere usar o KubernetesPodOperator
. Esse operador permite que você defina pods do Kubernetes e execute os pods em outros clusters.
Como adiciono pacotes personalizados não PyPI ou binários?
É possível instalar pacotes hospedados em repositórios de pacotes particulares.
Como faço para transmitir argumentos uniformemente para um DAG e as tarefas dele?
Use o suporte integrado do Airflow para modelos Jinja para transmitir argumentos que podem ser usados em campos com modelo.
Quando ocorre a substituição do modelo?
A substituição de modelo ocorre nos workers do Airflow pouco antes da função pre_execute
de um operador ser chamada. Na prática, isso significa que os modelos não são
substituídos até um pouco antes de uma tarefa ser executada.
Como sei quais argumentos do operador são compatíveis com a substituição do modelo?
Os argumentos do operador compatíveis com a substituição de modelos Jinja2 são explicitamente marcados como tal.
Procure o campo template_fields
na definição do operador, que contém uma lista de nomes de argumentos que passam por substituição de modelo.
Por exemplo, consulte
o BashOperator
, que é compatível com a modelagem dos
argumentos bash_command
e env
.
A seguir
- Como resolver problemas de DAGs
- Solução de problemas do programador
- Operadores do Google
- Operadores do Google Cloud
- Tutorial do Apache Airflow (em inglês)