Migrar ambientes para o Airflow 2

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Nesta página, explicamos como transferir DAGs, dados e configurações dos ambientes atuais do Airflow 1.10.* para ambientes com o Airflow 2 e versões mais recentes.

Observação: o Cloud Composer é compatível com transferências manuais de ambientes do Airflow 1.10.* para ambientes do Airflow 2. Não é possível fazer upgrade dos ambientes do Airflow 1.10.* para o Airflow 2. É possível fazer upgrade de ambientes com o Airflow 2 para versões mais recentes do Airflow normalmente.

Outros guias de migração

De	Até	Método	Guia
Cloud Composer 1, Airflow 2	Cloud Composer 2, Airflow 2	Lado a lado, usando snapshots	Guia de migração (snapshots)
Cloud Composer 1, Airflow 1	Cloud Composer 2 e Airflow 2	Lado a lado, usando snapshots	Guia de migração (instantâneos)
Cloud Composer 1 e Airflow 2	Cloud Composer 2, Airflow 2	Lado a lado, transferência manual	Guia de migração manual
Cloud Composer 1, Airflow 1	Cloud Composer 2 e Airflow 2	Transferência manual lado a lado	Guia de migração manual
Airflow 1	Airflow 2	Transferência manual lado a lado	Este guia (migração manual)

Upgrades lado a lado

O Cloud Composer conta com o script de transferência do banco de dados do Cloud Composer para migrar o banco de dados de metadados, os DAGs, os dados e os plug-ins dos ambientes do Cloud Composer com o Airflow 1.10.14 e o Airflow 1.10.15. para ambientes atuais do Cloud Composer com o Airflow 2.0.1 e versões mais recentes.

Esse é um caminho alternativo para o que descrevemos neste guia. Algumas partes deste guia ainda se aplicam ao usar o script fornecido. Por exemplo, talvez você queira verificar a compatibilidade dos DAGs com o Airflow 2 antes de migrá-los ou garantir que execuções simultâneas de DAGs não aconteçam e não haja execuções de DAG extras ou ausentes.

Antes de começar

Antes de começar, analise as mudanças que o AirFlow 2 faz nos ambientes do Cloud Composer.

Programador de alta disponibilidade

É possível usar mais de um programador do Airflow no ambiente. Basta definir o número de programadores ao criar um ambiente ou atualizar um ambiente.

Observação: é possível substituir as opções de configuração do programador do Airflow 2. Assim você configura o comportamento do programador, mas não ativa o programador de alta disponibilidade.

Executor do Kubernetes + Celery

Executor Celery+Kubernetes do Airflow 2 é compatível com o Cloud Composer 3.

Alterações importantes

O Airflow 2 introduz muitas alterações importantes, sendo que algumas estão causando falhas:

• Os DAGs atuais do Airflow 1.10.* não funcionam com o Airflow 2. Eles precisam ser testados e possivelmente ajustados.
• Operadores, transferências, hooks foram migrados para os pacotes dos provedores. As instruções de importação nos DAGs precisam usar novos pacotes de provedores. As instruções de importação antigas podem não funcionar mais no Airflow 2.
• Algumas configurações do Airflow 1.10.* podem não ser mais compatíveis, porque o Airflow 2 não aceita mais determinadas opções de configuração.
• Alguns pacotes PyPI personalizados podem ser incompatíveis com a nova versão do Airflow ou do Python.
• A interface do Airflow com controle de acesso é a UI padrão do Airflow 2. O Airflow 2 não é compatível com outros tipos de IU do Airflow.
• A API REST experimental foi substituída pela API Airflow estável. A API REST experimental está desativada por padrão no Airflow 2.
• Outras mudanças importantes no Airflow 2.0.0
• Outras mudanças importantes no Airflow 2.0.1

Diferenças entre ambientes com o Airflow 2 e o Airflow 1.10.*

Principais diferenças entre os ambientes do Cloud Composer com o Airflow 1.10.* e os ambientes com o Airflow 2:

• Ambientes com o Airflow 2 usam Python 3.8. Essa é uma versão mais recente do que a usada em ambientes do Airflow 1.10.*. Python 2, Python 3.6, Python 3.7 não são compatíveis.
• O Airflow 2 usa um formato de CLI diferente. O Cloud Composer fica compatível com o novo formato em ambientes com o Airflow 2 por meio do comando gcloud composer environments run.
• Os pacotes PyPI pré-instalados são diferentes nos ambientes do Airflow 2. Para ver os pacotes PyPI pré-instalados, consulte a Lista de versões do Cloud Composer.
• A serialização de DAG está sempre ativada no Airflow 2. Como resultado, o carregamento assíncrono de DAGs não é mais necessário e está indisponível no Airflow 2. Por isso, não é possível configurar os parâmetros [core]store_serialized_dags e [core]store_dag_code no Airflow 2, e as tentativas de defini-los serão relatadas como erros.
• Os plug-ins do servidor da Web do Airflow não são compatíveis. Isso não afeta os plug-ins do programador ou do worker, incluindo os operadores e sensores do Airflow.
• Nos ambientes do Airflow 2, o papel de usuário padrão do Airflow é Op. Para ambientes com o Airflow 1.10.*, o papel padrão é Admin.

Etapa 1: verificar a compatibilidade com o Airflow 2

Para verificar possíveis conflitos com o Airflow 2, consulte a seção "Fazer upgrade para do Airflow 2.0+, na seção sobre como fazer upgrade dos DAGs.

Um problema comum que você pode encontrar está relacionado à importação incompatível caminhos de rede. Para saber mais sobre como resolver esse problema de compatibilidade, no guia "Como fazer upgrade para o Airflow 2.0 e versões posteriores", consulte a seção sobre provedores de backport.

Etapa 2: criar um ambiente do Airflow 2, transferir substituições de configuração e variáveis de ambiente

Crie um ambiente do Airflow 2 e transfira as substituições de configuração e variáveis de ambiente:

1. Siga as etapas para criar um ambiente. Antes de criar um ambiente, especifique também as substituições de configuração e as variáveis de ambiente, conforme explicado mais adiante.

2. Ao selecionar uma imagem, escolha uma imagem com o Airflow 2.

3. Transfira manualmente os parâmetros de configuração do seu ambiente do Airflow 1.10.* para o novo ambiente do Airflow 2.

   Console

   1. Ao criar um ambiente, expanda a seção Redes, substituições de configurações do Airflow e recursos adicionais.

   2. Em Substituições da configuração do Airflow, clique em Adicionar subtituição da configuração do Airflow.

   3. Copie todas as substituições de configuração do ambiente do Airflow 1.10.*.

      Algumas opções de configuração têm nome e seção diferentes no Airflow 2. Para saber mais, consulte Alterações de configuração.

   4. Em Variáveis de ambiente, clique em Adicionar variável de ambiente.

   5. Copie todas as variáveis do ambiente do Airflow 1.10.*.

   6. Clique em Criar para criar um ambiente.

Etapa 3: instalar pacotes PyPI no ambiente do Airflow 2

Após a criação do ambiente do Airflow 2, instale pacotes PyPl:

Console

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

   Acessar "Ambientes"

2. Selecione o ambiente do Airflow 2.

3. Acesse a guia Pacotes PyPI e clique em Editar.

4. Copie os requisitos do pacote PyPI do ambiente do Airflow 1.10.*. Clique em Salvar e aguarde a atualização do ambiente.

   Como os ambientes do Airflow 2 usam um conjunto diferente de pacotes pré-instalados e outra versão do Python, talvez ocorram conflitos de pacote PyPI difíceis de resolver. Uma maneira de diagnosticar problemas de dependência de pacotes é instalar pacotes em um pod de worker do Airflow para verificar se há erros de pacote PyPI.

Etapa 4: transferir variáveis e pools para o Airflow 2

O Airflow 1.10.* é compatível com a exportação de variáveis e pools para arquivos JSON. Em seguida, importe esses arquivos para o ambiente do Airflow 2.

Se você tiver pools personalizados diferentes de default_pool, basta transferi-los. Caso contrário, ignore os comandos que exportam e importam pools.

Observação: os comandos da CLI do Airflow usados nesta etapa operam em arquivos locais dos workers do Airflow. Para fazer upload ou download dos arquivos, use a pasta /data no bucket do Cloud Storage do ambiente. Essa pasta é sincronizada com o diretório /home/airflow/gcs/data/ nos workers do Airflow. Nos comandos da CLI do Airflow, especifique /home/airflow/gcs/data/ no parâmetro FILEPATH.

gcloud

1. Exporte variáveis do ambiente do Airflow 1.10.*:

   gcloud composer environments run AIRFLOW_1_ENV \
       --location AIRFLOW_1_LOCATION \
        variables -- -e /home/airflow/gcs/data/variables.json
   

   Substitua:

   • AIRFLOW_1_ENV pelo nome do ambiente do Airflow 1.10.*.;
   • AIRFLOW_1_LOCATION pela região em que o ambiente está localizado;

2. Exporte pools do ambiente do Airflow 1.10.*:

   gcloud composer environments run AIRFLOW_1_ENV \
       --location AIRFLOW_1_LOCATION \
        pool -- -e /home/airflow/gcs/data/pools.json
   

3. Veja o URI do bucket do ambiente do Airflow 2.

   1. Execute este comando:

      gcloud composer environments describe AIRFLOW_2_ENV \
          --location AIRFLOW_2_LOCATION \
           --format="value(config.dagGcsPrefix)"
      

      Substitua:

      • AIRFLOW_2_ENV pelo nome do ambiente do Airflow 2;
      • AIRFLOW_2_LOCATION pela região em que o ambiente está localizado;

   2. Na saída, remova a pasta /dags. O resultado é o URI de bucket do ambiente do Airflow 2.

      Por exemplo, altere gs://us-central1-example-916807e1-bucket/dags para gs://us-central1-example-916807e1-bucket.

4. Transfira arquivos JSON com variáveis e pools para o ambiente do Airflow 2:

   gcloud composer environments storage data export \
       --destination=AIRFLOW_2_BUCKET/data \
       --environment=AIRFLOW_1_ENV \
       --location=AIRFLOW_1_LOCATION \
       --source=variables.json
   
   gcloud composer environments storage data export \
       --destination=AIRFLOW_2_BUCKET/data \
       --environment=AIRFLOW_1_ENV \
       --location=AIRFLOW_1_LOCATION \
       --source=pools.json
   

   Substitua AIRFLOW_2_BUCKET pelo URI do bucket do ambiente do Airflow 2 recebido na etapa anterior.

5. Importe variáveis e pools para o Airflow 2:

   gcloud composer environments run \
       AIRFLOW_2_ENV \
       --location AIRFLOW_2_LOCATION \
       variables import \
       -- /home/airflow/gcs/data/variables.json
   
   gcloud composer environments run \
       AIRFLOW_2_ENV \
       --location AIRFLOW_2_LOCATION \
       pools import \
       -- /home/airflow/gcs/data/pools.json
   

6. Verifique se as variáveis e os pools foram importados:

   gcloud composer environments run \
       AIRFLOW_2_ENV \
       --location AIRFLOW_2_LOCATION \
       variables list
   
   gcloud composer environments run \
       AIRFLOW_2_ENV \
       --location AIRFLOW_2_LOCATION \
       pools list
   

7. Remova arquivos JSON dos buckets:

   gcloud composer environments storage data delete \
       variables.json \
       --environment=AIRFLOW_2_ENV \
       --location=AIRFLOW_2_LOCATION
   
   gcloud composer environments storage data delete \
       pools.json \
       --environment=AIRFLOW_2_ENV \
       --location=AIRFLOW_2_LOCATION
   
   gcloud composer environments storage data delete \
       variables.json \
       --environment=AIRFLOW_1_ENV \
       --location=AIRFLOW_1_LOCATION
   
   gcloud composer environments storage data delete \
       pools.json \
       --environment=AIRFLOW_1_ENV \
       --location=AIRFLOW_1_LOCATION
   

Etapa 5: transferir outros dados do bucket do ambiente do Airflow 1.10.*

gcloud

1. Transfira plug-ins para o ambiente do Airflow 2. Para fazer isso, exporte plug-ins do bucket do ambiente do Airflow 1.10.* para a pasta /plugins no bucket do ambiente do Airflow 2:

   gcloud composer environments storage plugins export \
       --destination=AIRFLOW_2_BUCKET/plugins \
       --environment=AIRFLOW_1_ENV \
       --location=AIRFLOW_1_LOCATION
   

2. Verifique se a pasta /plugins foi importada com sucesso:

   gcloud composer environments storage plugins list \
       --environment=AIRFLOW_2_ENV \
       --location=AIRFLOW_2_LOCATION
   

3. Exporte a pasta /data do ambiente do Airflow 1.10.* para o ambiente do Airflow 2:

       gcloud composer environments storage data export \
           --destination=AIRFLOW_2_BUCKET/data \
           --environment=AIRFLOW_1_ENV \
           --location=AIRFLOW_1_LOCATION
   

4. Verifique se a pasta /data foi importada com sucesso:

   gcloud composer environments storage data list \
       --environment=AIRFLOW_2_ENV \
       --location=AIRFLOW_2_LOCATION
   

Etapa 6: transferir conexões e usuários

O Airflow 1.10.* não é compatível com a exportação de usuários e conexões. Para transferir usuários e conexões, crie manualmente novas contas de usuário e conexões no ambiente do Airflow 2.

Importante: os ambientes do Cloud Composer têm muitas conexões padrão. Não transfira essas conexões porque os ambientes do Airflow 2 já as têm.

gcloud

1. Para ver uma lista de conexões no ambiente do Airflow 1.10.*, execute:

   gcloud composer environments run AIRFLOW_1_ENV \
       --location AIRFLOW_1_LOCATION \
        connections -- --list
   

2. Para criar uma nova conexão no ambiente do Airflow 2, execute o comando connections da CLI do Airflow com o gcloud. Exemplo:

   gcloud composer environments run \
       AIRFLOW_2_ENV \
       --location AIRFLOW_2_LOCATION \
       connections add \
       -- --conn-host postgres.example.com \
       --conn-port 5432 \
       --conn-type postgres \
       --conn-login example_user \
       --conn-password example_password \
       --conn-description "Example connection" \
       example_connection
   

3. Para ver uma lista de usuários no ambiente do Airflow 1.10.*:

   1. Abra a interface da Web do Airflow para o ambiente do Airflow 1.10.*.

   2. Acesse Administrador > Usuários.

4. Para criar uma nova conta de usuário no ambiente do Airflow 2, execute o comando users create da CLI do Airflow com o gcloud. Exemplo:

   gcloud composer environments run \
       AIRFLOW_2_ENV \
       --location AIRFLOW_2_LOCATION \
       users create \
       -- --username example_username \
       --firstname Example-Name \
       --lastname Example-Surname \
       --email [e-mail] \
       --use-random-password \
       --role Admin
   

Etapa 7: verificar se os DAGs estão prontos para o Airflow 2

Antes de transferir DAGs para o ambiente do Airflow 2, verifique se:

1. Seus DAGs são executados com sucesso e não há problemas de compatibilidade restantes.

2. Seus DAGs usam instruções de importação corretas.

   Por exemplo, a nova instrução de importação para BigQueryCreateDataTransferOperator pode ser assim:

   from airflow.providers.google.cloud.operators.bigquery_dts \
       import BigQueryCreateDataTransferOperator
   

3. Os DAGs estão atualizados para o Airflow 2. Essa mudança é compatível com o Airflow 1.10.14 e versões mais recentes.

Etapa 8: transferir DAGs para o ambiente do Airflow 2

Aviso: nesta etapa, explicamos como evitar possíveis problemas com execuções de DAGs duplicadas, ausentes, incorretas e inválidas ao transferir DAGs entre ambientes.

Os seguintes problemas podem acontecer quando você transfere DAGs entre ambientes:

• Se um DAG estiver ativado (não pausado) nos dois ambientes, cada ambiente executará uma cópia própria do DAG, conforme programado. Isso pode gerar execuções simultâneas de DAGs para os mesmos dados e tempo de execução.

• Devido à atualização do DAG, o Airflow programa execuções extras de DAGs, começando pela data de início especificada nos DAGs. Isso acontece porque a nova instância do Airflow não considera o histórico de execuções do DAG no ambiente da versão 1.10.*. Isso pode levar a um grande número de execuções de DAGs programadas a partir da data de início especificada.

Impedir execuções simultâneas de DAGs

No ambiente do Airflow 2, substitua a opção de configuração dags_are_paused_at_creation do Airflow. Depois que você fizer essa alteração, todos os novos DAGs serão pausados por padrão.

Seção	Chave	Valor
core	dags_are_paused_at_creation	True

Observação: é possível remover essa substituição depois de transferir todos os DAGs para o ambiente do Airflow 2.

Evitar execuções extras ou ausentes de DAGs

Especifique uma nova data de início estática nos DAGs que você transferiu para o ambiente do Airflow 2.

Para evitar lacunas e sobreposições nas datas de execução, a primeira execução do DAG precisa acontecer no ambiente do Airflow 2 na próxima ocorrência do intervalo de programação. Para fazer isso, defina a nova data de início no DAG como anterior à data da última execução no ambiente do Airflow 1.10.*.

Por exemplo, se o DAG for executado às 15h, 17h e 21h todos os dias no ambiente do Airflow 1.10.*, a última execução do DAG aconteceu às 15h e você pretende transferir o DAG às 15h15, a data de início do ambiente do Airflow 2 pode ser hoje às 14h45. Depois de ativar o DAG no ambiente do Airflow 2, ele programará uma execução do DAG para às 17h.

Outro exemplo: se o DAG for executado à 00h todos os dias no ambiente do Airflow 1.10.*, a última execução do DAG aconteceu às 00h de 26 de abril de 2021 e você planeja transferir o DAG às 13h de 26 de abril de 2021, a data de início do ambiente do Airflow 2 pode ser às 23h45 de 25 de abril de 2021. Depois de ativar o DAG no ambiente do Airflow 2, ele programará uma execução do DAG para 00h do dia 27 de abril de 2021.

Transferir os DAGs um por um para o ambiente do Airflow 2

Siga este procedimento para transferir cada DAG:

1. Verifique se a nova data de início no DAG está configurada conforme descrito na seção anterior.

2. Faça upload do DAG atualizado para o ambiente do Airflow 2. Esse DAG está pausado no ambiente do Airflow 2 devido à substituição da configuração. Portanto, nenhuma execução de DAG ainda está programada.

3. Na interface da Web do Airflow, acesse DAGs e verifique se há erros de sintaxe de DAG relatados.

   Observação: a interface da Web do Airflow não exibe corretamente os DAGs localizados em arquivos ZIP. Esse comportamento é uma limitação conhecida no Airflow.

4. No momento em que você planeja transferir o DAG:

   1. Pause o DAG no ambiente do Airflow 1.10.*.

   2. Cancele a pausa do DAG no ambiente do Airflow 2.

   3. Verifique se a nova execução do DAG está programada para o horário correto.

   4. Aguarde a execução do DAG acontecer no ambiente do Airflow 2 e verifique se foi bem-sucedida.

5. Se a execução do DAG for bem-sucedida:

   • Se a execução do DAG for bem-sucedida, será possível continuar e usar o DAG no ambiente do Airflow 2. Por fim, considere excluir a versão do Airflow 1.10.* do DAG.

   • Se a execução do DAG falhar, tente resolver problemas do DAG até que ele seja executado com sucesso no Airflow 2.

     Se necessário, sempre é possível recorrer à versão 1.10.* do Airflow do DAG:

     1. Pause o DAG no ambiente do Airflow 2.

     2. Cancele a pausa do DAG no ambiente do Airflow 1.10.*. Assim você programa uma nova execução do DAG para a mesma data e hora que a execução do DAG com falha.

     3. Quando tudo estiver pronto para continuar com a versão do Airflow 2 do DAG, ajuste a data de início, faça upload da nova versão do DAG para o ambiente do Airflow 2 e repita o procedimento.

Etapa 9: monitorar o ambiente do Airflow 2

Depois de transferir todos os DAGs e a configuração para o ambiente do Airflow 2, monitore-os em busca de possíveis problemas, execuções com falha do DAG e integridade geral do ambiente. Se o ambiente do Airflow 2 for executado sem problemas por um período suficiente, será possível remover o ambiente do Airflow 1.10.*.

A seguir

• Como resolver problemas de DAGs
• Como resolver problemas da criação de ambientes
• Solução de problemas de atualizações do ambiente
• Como usar pacotes de backport