Testar, sincronizar e implantar seus DAGs do GitHub

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Neste guia, explicamos como criar um pipeline de CI/CD para testar, sincronizar e implantar DAGs do GitHub no ambiente do Cloud Composer repositório de dados.

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

Visão geral do pipeline de CI/CD

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

.

O pipeline de CI/CD que, para testar, sincronizar e implantar DAGs, tem a etapas a seguir:

1. Você faz uma alteração 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 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 da repositório de dados.

5. O Cloud Build sincroniza o desenvolvimento ambiente do Cloud Composer com essas novas alterações.

6. Você verifica se o DAG se comporta conforme o esperado no seu desenvolvimento de nuvem.

7. Se o DAG funcionar conforme o esperado, faça o upload dele para a produção ambiente do Cloud Composer.

Objetivos

Objetivo:

• Executar uma verificação automatizada de pré-envio usando o Cloud Build. Essa verificação é executada e testes de unidade para um DAG.
• Sincronize os DAGs no ambiente de desenvolvimento do Cloud Composer com DAGs no seu repositório do GitHub.

Antes de começar

• Este guia pressupõe que você esteja trabalhando com duas instâncias Ambientes do Cloud Composer: um ambiente de desenvolvimento e um ambiente de produção.

  Neste guia, você configura apenas um pipeline de CI/CD para seu ambiente de desenvolvimento. Certifique-se de que o ambiente usado não é um ambiente de produção.

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

  O exemplo de pipeline de CI/CD o conteúdo de um repositório de exemplo. DAGs e testes são armazenadas no diretório dags/, com arquivos de requisitos, as restrições e os arquivos de configuração do Cloud Build armazenados no nível superior. O utilitário de sincronização do DAG e os requisitos dele estão localizados no 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 a unidade e testes para os DAGs.

Adicionar testes de unidade

Se ainda não fez isso, crie testes de unidade para seus DAGs. Salve esses testes junto com DAGs no seu repositório, cada um com o sufixo _test. Por exemplo, o teste para o DAG em example_dag.py é example_dag_test.py. Estes 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 configura o job do Cloud Build para verificações de pré-envio. Nele, há três etapas:

1. Instale as dependências necessárias para os DAGs.
2. Instale as dependências necessárias para os testes de unidade.
3. Execute os testes do 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 Python do Docker Hub e você pode especifique a tag no campo name de cada etapa. Se você estiver usando python como entrypoint, verifique se entrypoint corresponde à versão especificada na tag Caso contrário, o padrão entrypoint será latest.

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

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

• Nome: test-dags

• Evento: Solicitação de envio

• Origem - Repositório: escolha o repositório.

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

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

  Atenção: se você configurar o controle de comentários como não obrigatória, os usuários pode enviar uma solicitação pull que aciona o build. Para repositórios maiores ou repositórios com colaboradores não confiáveis, exigindo o controle de comentários recomendado.

• 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 do DAG e adicionar o script utilitário deles

Em seguida, configure um job do Cloud Build que execute um script utilitário de DAGs. O script de utilitário neste job sincroniza seus DAGs com o Ambiente do Cloud Composer depois de mesclados com a ramificação principal no seu repositório.

Observação: o Cloud Composer analisa todos os DAGs em um intervalo repetido. Isso acontece mesmo que os DAGs não tenham mudado desde a última análise. Por isso esse utilitário faz upload de todos os DAGs no Cloud Composer, e não apenas e aqueles que mudaram.

Adicionar o script de utilitário dos DAGs

Adicione o script de utilitário do DAG ao seu repositório. Esse script utilitário copia todos os arquivos DAG no diretório dags/ do em um diretório temporário, ignorando todos os arquivos Python que não sejam DAG. 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/ na sua bucket do ambiente do Cloud Composer.

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 gsutil -m cp -r 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

Em seu repositório, crie um arquivo YAML chamado add-dags-to-composer.cloudbuild.yaml que configura o Cloud Build para sincronizar os DAGs. Nela, há duas etapas:

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

2. Execute o script utilitário para sincronizar os DAGs no repositório com o ambiente do Cloud Composer.

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 Python do Docker Hub e você pode especifique a tag no campo name de cada etapa. Se você estiver usando python como entrypoint, verifique se entrypoint corresponde à versão especificada na tag Caso contrário, o padrão entrypoint será latest.

Crie o gatilho do Cloud Build

Siga as instruções em Como criar repositórios do GitHub. guia para criar um gatilho baseado em 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 o repositório.

• Origem: ramificação de base: ^main$ (mude main para o nome da na ramificação de base do 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)

Em "Configuração avançada", adicione dois variáveis de substituição:

• _DAGS_DIRECTORY: o diretório em que os DAGs estão localizados no 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 Diretório dags/ no seu Cloud Composer de desenvolvimento de nuvem. Omita o prefixo gs://. Por exemplo: us-central1-example-env-1234ab56-bucket.

.

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

Testar o pipeline de CI/CD

Nesta seção, siga um fluxo de desenvolvimento do DAG que utiliza seu novo criou gatilhos do Cloud Build.

Executar um job de pré-envio

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

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. Seu ambiente do Cloud Composer pode ter uma de pacotes PyPI e pacotes pré-instalados. Recomendamos executar um ambiente local do Airflow durante o desenvolvimento da DAGs.

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

.

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

validar se o DAG funciona no ambiente de desenvolvimento do Cloud Composer

Depois que sua solicitação de envio for aprovada, mescle-a à ramificação principal. Use o console do Google Cloud para ver os resultados da versão. Se você tiver muitos gatilhos do Cloud Build, é possível filtrar os builds pelo nome do gatilho add-dags-to-composer:

Depois que o job de sincronização do Cloud Build for bem-sucedido, o DAG sincronizado será exibido no ambiente de desenvolvimento do Cloud Composer. Lá, é possível validar se o DAG funciona como esperado.

Adicionar o DAG ao ambiente de produção

Adicione o DAG manualmente à produção depois que ele for executado conforme o esperado. de nuvem. Para isso, faça upload do arquivo DAG para o diretório dags/ no Cloud Composer de produção bucket do seu ambiente de execução.

Se o job de sincronização do DAG falhar ou não funcionar como esperado no DAG de desenvolvimento do Cloud Composer, consulte Como resolver falhas de build.

Como resolver falhas no build

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

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

Na sua solicitação de envio, clique em Detalhes e escolha Confira mais detalhes no Google Cloud Build para acessar os registros de build na console do Google Cloud. Use esses registros para depurar o problema da DAG. Depois de resolver os problemas, confirme a correção e envie para seu ramificação. A verificação de pré-envio é executada novamente, e é possível continuar iterando usando os registros como uma ferramenta de depuração.

E se o job de sincronização do DAG falhar?

Use o console do Google Cloud para ver os resultados da versão. Se você tiver muitos gatilhos do Cloud Build, é possível filtrar 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 canais de suporte.

E se o DAG não funcionar corretamente no ambiente do Cloud Composer?

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

• Reverter a solicitação de envio com as alterações que interromperam o DAG para restaurá-lo ao estado imediatamente antes das alterações. Isso também reverte todos os outros arquivos na solicitação de envio.
• Crie uma nova solicitação de envio para reverter manualmente as alterações 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 do DAG.

A seguir

• Executar ambientes locais do Airflow
• Gravar DAGs
• Acionar DAGs
• Testar DAGs