Testar, sincronizar e implantar seus DAGs do GitHub

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

Este guia explica como criar um pipeline de CI/CD para testar, sincronizar e implantar DAGs no ambiente do Airflow gerenciado do seu repositório do GitHub.

Se você quiser apenas sincronizar dados de outros serviços, consulte Transferir dados de outros serviços.

Visão geral do pipeline de CI/CD

Figura 1. Diagrama de arquitetura mostrando as etapas do fluxo (clique para ampliar)

O pipeline de CI/CD para testar, sincronizar e implantar DAGs tem as seguintes etapas:

1. Você faz uma mudança em um DAG e a envia para uma ramificação de desenvolvimento no seu repositório.

2. Você abre uma solicitação de envio na ramificação principal do seu repositório.

3. O Cloud Build executa testes de unidade para verificar se o DAG é válido.

4. Sua solicitação de envio é aprovada e mesclada à ramificação principal do seu repositório.

5. O Cloud Build sincroniza o ambiente do Airflow gerenciado de desenvolvimento com essas novas mudanças.

6. Você verifica se o DAG funciona conforme o esperado no ambiente de desenvolvimento

7. Se o DAG funcionar conforme o esperado, faça upload dele para o ambiente do Airflow gerenciado de produção.

Objetivos

• Executar uma verificação automatizada de pré-envio usando o Cloud Build. Essa verificação executa testes de unidade para um DAG.
• Sincronizar DAGs no ambiente do Serviço gerenciado para Apache Airflow de desenvolvimento com DAGs no seu repositório do GitHub.

Antes de começar

• Este guia pressupõe que você está trabalhando com dois ambientes idênticos do Airflow gerenciado: um ambiente de desenvolvimento e um ambiente de produção.

  Para fins deste guia, você está configurando um pipeline de CI/CD apenas para o ambiente de desenvolvimento. Verifique se o ambiente usado não é de produção.

• Este guia pressupõe que você tenha seus DAGs e testes armazenados em um repositório do GitHub.

  O pipeline de CI/CD de exemplo demonstra o conteúdo de um repositório de exemplo. Os DAGs e testes são armazenados no diretório dags/, com arquivos de requisitos, o arquivo de restrições e arquivos de configuração do Cloud Build armazenados no nível superior. O utilitário de sincronização de DAGs e os requisitos dele estão localizados no diretório utils.

Criar um job de verificação de pré-envio e testes de unidade

O primeiro job do Cloud Build executa uma verificação de pré-envio, que executa testes de unidade para seus DAGs.

Adicionar testes de unidade

Se ainda não tiver feito isso, crie testes de unidade para seus DAGs. Salve esses testes junto com os DAGs no seu repositório, cada um com o sufixo _test. Por exemplo, o arquivo de teste do DAG em example_dag.py é example_dag_test.py. Esses são os testes executados como uma verificação de pré-envio no seu repositório.

Criar configuração YAML do Cloud Build para a verificação de pré-envio

No seu repositório, crie um arquivo YAML chamado test-dags.cloudbuild.yaml que configure o job do Cloud Build para verificações de pré-envio. Nele, há três etapas:

1. Instalar as dependências necessárias para seus DAGs.
2. Instalar as dependências necessárias para seus testes de unidade.
3. Executar os testes de DAG.

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

Observação: o Cloud Build extrai a imagem do Python do Docker Hub, e você pode especificar a tag no campo name de cada etapa. Se você estiver usando python como entrypoint, verifique se o entrypoint corresponde à versão especificada na tag. Caso contrário, o entrypoint será definido como latest por padrão.

Criar o gatilho de build do Cloud Build para a verificação de pré-envio

Siga o guia Como criar repositórios do GitHub para criar um gatilho baseado em um app do GitHub com as seguintes configurações:

• Nome: test-dags

• Evento: solicitação de envio

• Origem - Repositório: escolha seu repositório

• Origem - Ramificação de base: ^main$ (mude main para o nome da ramificação de base do seu repositório, se necessário)

• Origem - Controle de comentários: não obrigatório

  Cuidado: definir o controle de comentários como não obrigatório significa que qualquer usuário pode enviar uma solicitação de envio que acione o build. Para repositórios maiores ou com colaboradores não confiáveis, é recomendável exigir o controle de comentários.

• Configuração do build - Arquivo de configuração do Cloud Build: /test-dags.cloudbuild.yaml (o caminho para o arquivo de build)

Criar um job de sincronização de DAGs e adicionar o script de utilitário de DAGs

Em seguida, configure um job do Cloud Build que execute um script de utilitário de DAGs. O script de utilitário nesse job sincroniza seus DAGs com o ambiente do Airflow gerenciado depois que eles são mesclados à ramificação principal no seu repositório.

**Observação** :o Airflow gerenciado analisa todos os DAGs em um intervalo repetido. Isso acontece mesmo que os DAGs não tenham mudado desde a última análise. Devido a esse comportamento, esse utilitário faz upload de todos os DAGs para o Airflow gerenciado, não apenas aqueles que foram alterados.

Adicionar o script de utilitário de DAGs

Adicione o script de utilitário de DAGs ao seu repositório. Esse script de utilitário copia todos os arquivos DAG no diretório dags/ do seu repositório para um diretório temporário, ignorando todos os arquivos Python que não são DAGs. Em seguida, o script usa a biblioteca de cliente do Cloud Storage para fazer upload de todos os arquivos desse diretório temporário para o diretório dags/ no bucket do ambiente do Airflow gerenciado.

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage


def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)


def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gcloud storage cp --recursive on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

Criar configuração YAML do Cloud Build para sincronizar DAGs

No seu repositório, crie um arquivo YAML chamado add-dags-to-composer.cloudbuild.yaml que configure o job do Cloud Build para sincronizar DAGs. Nele, há duas etapas:

1. Instalar as dependências necessárias para o script de utilitário de DAGs.

2. Executar o script de utilitário para sincronizar os DAGs no seu repositório com o ambiente do Airflow gerenciado.

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

Observação: o Cloud Build extrai a imagem do Python do Docker Hub, e você pode especificar a tag no campo name de cada etapa. Se você estiver usando python como entrypoint, verifique se o entrypoint corresponde à versão especificada na tag. Caso contrário, o entrypoint será definido como latest por padrão.

Criar o gatilho de build do Cloud Build

Siga o guia Como criar repositórios do GitHub para criar um gatilho baseado em um app do GitHub com as seguintes configurações:

• Nome: add-dags-to-composer

• Evento: enviar por push para uma ramificação

• Origem - Repositório: escolha seu repositório

• Origem - Ramificação de base: ^main$ (mude main para o nome da ramificação de base do seu repositório, se necessário)

• Origem - Filtro de arquivos incluídos (glob): dags/**

• Configuração do build - Arquivo de configuração do Cloud Build: /add-dags-to-composer.cloudbuild.yaml (o caminho para o arquivo de build)

Na configuração avançada, adicione duas variáveis de substituição:

• _DAGS_DIRECTORY: o diretório em que os DAGs estão localizados no seu repositório. Se você estiver usando o repositório de exemplo deste guia, ele será dags/.

• _DAGS_BUCKET - o bucket do Cloud Storage que contém o dags/ diretório no ambiente do Airflow gerenciado de desenvolvimento. Omita o prefixo gs://. Por exemplo: us-central1-example-env-1234ab56-bucket.

Cuidado: se você não remover o sufixo dags/ do nome do bucket do Cloud Storage na variável de substituição _DAGS_BUCKET, o job de sincronização de DAGs vai falhar com um erro NotFound. Isso acontece porque, no Cloud Storage, dags/ é uma pasta, que é considerada parte do nome do objeto, não parte do nome do bucket.

Testar o pipeline de CI/CD

Nesta seção, siga um fluxo de desenvolvimento de DAGs que utiliza os gatilhos do Cloud Build recém-criados.

Executar um job de pré-envio

Crie uma solicitação de envio para sua ramificação principal para testar o build. Localize a verificação de pré-envio na página. Clique em Detalhes e escolha Ver mais detalhes no Google Cloud Build para conferir os registros de build no Google Cloud console.

Importante: o Cloud Build executa testes de unidade com pacotes especificados em requirements.txt e requirements-test.txt, que incluem uma versão pública do Airflow. O ambiente do Airflow gerenciado pode ter um conjunto diferente de pacotes PyPI e pacotes pré-instalados. Recomendamos executar um ambiente local do Airflow ao desenvolver seus DAGs.

Figura 2. Captura de tela do status da verificação de pré-envio do Cloud Build no GitHub (clique para ampliar)

Se a verificação de pré-envio falhou, consulte Como resolver falhas de build.

Validar se o DAG funciona no ambiente de desenvolvimento

Depois que a solicitação de envio for aprovada, mescle-a à ramificação principal. Use o Google Cloud console para conferir os resultados do build. Se você tiver muitos gatilhos do Cloud Build, filtre os builds pelo nome do gatilho add-dags-to-composer.

Depois que o job de sincronização do Cloud Build for concluído, o DAG sincronizado vai aparecer no ambiente do Airflow gerenciado de desenvolvimento. Nele, você pode validar se o DAG funciona conforme o esperado.

Adicionar o DAG ao ambiente de produção

Depois que o DAG funcionar conforme o esperado, adicione-o manualmente ao ambiente de produção. Para fazer isso, faça upload do arquivo DAG para o diretório dags/ no bucket do ambiente do Airflow gerenciado de produção.

Se o job de sincronização de DAGs falhou ou se o DAG não está se comportando conforme o esperado no seu ambiente do Airflow gerenciado de desenvolvimento, consulte Como resolver falhas de build.

Como resolver falhas de build

Esta seção explica como resolver cenários comuns de falha de build.

E se a verificação de pré-envio falhar?

Na solicitação de envio, clique em Detalhes e escolha Ver mais detalhes no Google Cloud Build para conferir os registros de build no Google Cloud console. Use esses registros para ajudar a depurar o problema com o seu DAG. Depois de resolver os problemas, faça commit da correção e envie para sua ramificação. A verificação de pré-envio é executada novamente, e você pode continuar a iterar usando os registros como uma ferramenta de depuração.

E se o job de sincronização de DAGs falhar?

Use o Google Cloud console para conferir os resultados do build. Se você tiver muitos gatilhos do Cloud Build, filtre os builds pelo nome do gatilho add-dags-to-composer. Examine os registros do job de build e resolva os erros. Se precisar de mais ajuda para resolver os erros, utilize os canais de suporte.

E se o DAG não funcionar corretamente no ambiente do Airflow gerenciado?

Se o DAG não funcionar conforme o esperado no ambiente do Airflow gerenciado de desenvolvimento, não promova o DAG manualmente para o ambiente do Airflow gerenciado de produção. Em vez disso, escolha uma das opções a seguir:

• Reverta a solicitação de pull com as mudanças que quebraram o DAG para restaurá-lo ao estado imediatamente anterior às mudanças. Isso também reverte todos os outros arquivos nessa solicitação de envio.
• Crie uma nova solicitação de envio para reverter manualmente as mudanças no DAG corrompido.
• Crie uma nova solicitação de envio para corrigir os erros no DAG.

Seguir qualquer uma dessas etapas aciona uma nova verificação de pré-envio e, após a mesclagem, o job de sincronização de DAGs.

A seguir

• Executar ambientes locais do Airflow
• Gravar DAGs
• Programar e acionar DAGs
• Testar DAGs