Executar um DAG do Apache Airflow no Cloud Composer 3 (CLI do Google Cloud)

Executar um DAG do Apache Airflow no Cloud Composer 3 (CLI do Google Cloud)

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Este guia de início rápido mostra como criar um ambiente do Cloud Composer e executar um DAG do Apache Airflow no Cloud Composer 3.

• Se você é novo no Airflow, consulte o tutorial de conceitos do Airflow na documentação do Apache Airflow para mais informações sobre conceitos, objetos e uso do Airflow.

• Se você quiser usar o console do Google Cloud em vez disso, consulte Como executar um DAG do Apache Airflow no Cloud Composer.

• Se você quiser criar um ambiente usando o Terraform, consulte Criar ambientes (Terraform).

Antes de começar

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Verifique se a cobrança está ativada para o seu projeto do Google Cloud.

6. Install the Google Cloud CLI.

7. To initialize the gcloud CLI, run the following command:

   gcloud init

8. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

9. Verifique se a cobrança está ativada para o seu projeto do Google Cloud.

10. Enable the Cloud Composer API:

    gcloud services enable composer.googleapis.com

11. Para conseguir as permissões necessárias a fim de concluir o guia de início rápido, peça ao administrador para conceder a você os seguintes papéis do IAM no projeto:

    • Para criar e gerenciar o ambiente do Cloud Composer:
      • Administrador de ambiente e de objetos do Storage (roles/composer.environmentAndStorageObjectAdmin)
      • Usuário da conta de serviço (roles/iam.serviceAccountUser)
    • Para conferir os registros: Visualizador de registros (roles/logging.viewer)

    Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

    Também é possível conseguir as permissões necessárias por meio de papéis personalizados ou de outros papéis predefinidos.

Criar um ambiente

Crie um novo ambiente chamado example-environment no us-central1. com a versão mais recente do Cloud Composer 3.

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-3-airflow-2.9.1-build.7

Criar um arquivo DAG

Um DAG do Airflow é uma coleção de tarefas organizadas que que você quer programar e executar. Os DAGs são definidos em arquivos Python padrão.

Neste guia, usamos um exemplo de DAG do Airflow definido no arquivo quickstart.py. O código Python nesse arquivo faz o seguinte:

1. Cria um DAG, composer_sample_dag. Esse DAG é executado todos os dias.
2. Executa uma tarefa, print_dag_run_conf. A tarefa imprime a configuração da execução do DAG usando o operador bash.

Salve uma cópia do arquivo quickstart.py na sua máquina local:

import datetime

from airflow import models
from airflow.operators import bash

# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

Faça upload do arquivo DAG para o bucket do ambiente

Todos os ambientes do Cloud Composer têm um bucket do Cloud Storage associado. O Airflow no Cloud Composer programa apenas os DAGs que estão na pasta /dags neste bucket.

Para programar o DAG, faça upload de quickstart.py da máquina local para a pasta /dags do ambiente:

Para fazer upload de quickstart.py com a Google Cloud CLI, execute o comando a seguir na pasta em que o arquivo quickstart.py está localizado:

gcloud composer environments storage dags import \
--environment example-environment --location us-central1 \
--source quickstart.py

Conferir o DAG

Depois de fazer upload do arquivo DAG, o Airflow faz o seguinte:

1. Analisa o arquivo DAG que você enviou. Pode levar alguns minutos para que o DAG fique disponível para o Airflow.
2. Adiciona o DAG à lista de DAGs disponíveis.
3. Executa o DAG de acordo com a programação fornecida no arquivo DAG.

Verifique se o DAG é processado sem erros e está disponível no Airflow abrindo a interface do DAG. A interface do DAG é a interface do Cloud Composer para visualização Informações do DAG no console do Google Cloud. O Cloud Composer também oferece acesso à interface do Airflow, que é uma interface da Web nativa do Airflow.

1. Aguarde cerca de cinco minutos para que o Airflow tenha tempo de processar o arquivo DAG que você enviou anteriormente e concluir a primeira execução do DAG (explicado mais tarde).

2. Execute o comando a seguir na Google Cloud CLI. Esse comando executa dags list Comando da CLI do Airflow que lista os DAGs na sua de nuvem.

   gcloud composer environments run example-environment \
   --location us-central1 \
   dags list
   

3. Verifique se o DAG composer_quickstart está listado na saída do comando.

   Exemplo de saída:

   Executing the command: [ airflow dags list ]...
   Command has been started. execution_id=d49074c7-bbeb-4ee7-9b26-23124a5bafcb
   Use ctrl-c to interrupt the command
   dag_id              | filepath              | owner            | paused
   ====================+=======================+==================+=======
   airflow_monitoring  | airflow_monitoring.py | airflow          | False
   composer_quickstart | dag-quickstart-af2.py | Composer Example | False
   

   Observação: o Cloud Composer usa o DAG airflow_monitoring para monitorar a integridade do ambiente.

Mais detalhes da execução do DAG

Uma única execução de um DAG é chamada de execução de DAG. O Airflow executa imediatamente uma execução de DAG para o exemplo, porque a data de início no arquivo DAG está definida como ontem. Assim, o Airflow alcança os DAGs especificados cronograma.

O DAG de exemplo contém uma tarefa, print_dag_run_conf, que executa o comando echo no console. Esse comando exibe metainformações sobre o DAG (identificador numérico da execução do DAG).

Execute o comando a seguir na Google Cloud CLI. Este comando lista as execuções de DAG para o DAG composer_quickstart:

gcloud composer environments run example-environment \
--location us-central1 \
dags list-runs -- --dag-id composer_quickstart

Exemplo de saída:

dag_id              | run_id                                      | state   | execution_date                   | start_date                       | end_date
====================+=============================================+=========+==================================+==================================+=================================
composer_quickstart | scheduled__2024-02-17T15:38:38.969307+00:00 | success | 2024-02-17T15:38:38.969307+00:00 | 2024-02-18T15:38:39.526707+00:00 | 2024-02-18T15:38:42.020661+00:00

Observação :é possível usar o comando tasks states-for-dag-run para conferir o status de tarefas individuais.

A CLI do Airflow não fornece um comando para visualizar os registros de tarefas. É possível usar outros métodos para acessar os registros de tarefas do Airflow: a interface do DAG do Cloud Composer, a interface do Airflow ou o Cloud Logging. Este guia mostra uma maneira de consultar o Cloud Logging para encontrar registros de uma execução específica de DAG.

Execute o comando a seguir na CLI do Google Cloud. Esse comando lê os registros do Cloud Logging para uma execução de DAG específica do DAG composer_quickstart. O O argumento --format formata a saída para que apenas o texto da mensagem de registro é exibido.

gcloud logging read \
--format="value(textPayload)" \
--order=asc \
"resource.type=cloud_composer_environment \
resource.labels.location=us-central1 \
resource.labels.environment_name=example-environment \
labels.workflow=composer_quickstart \
(labels.\"execution-date\"=\"RUN_ID\")"

Substitua:

• RUN_ID com o valor run_id da saída do comando tasks states-for-dag-run executado anteriormente. Por exemplo, 2024-02-17T15:38:38.969307+00:00.

Exemplo de saída:

...

Starting attempt 1 of 2
Executing <Task(BashOperator): print_dag_run_conf> on 2024-02-17
15:38:38.969307+00:00
Started process 22544 to run task

...

Running command: ['/usr/bin/bash', '-c', 'echo 115746']
Output:
115746

...

Command exited with return code 0
Marking task as SUCCESS. dag_id=composer_quickstart,
task_id=print_dag_run_conf, execution_date=20240217T153838,
start_date=20240218T153841, end_date=20240218T153841
Task exited with return code 0
0 downstream tasks scheduled from follow-on schedule check

Limpar

Para evitar cobranças na sua conta do Google Cloud pelos recursos usados nesta página, exclua o projeto do Google Cloud com esses recursos.

Exclua os recursos usados neste tutorial:

1. Exclua o ambiente do Cloud Composer:

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

      Acessar "Ambientes"

   2. Selecione example-environment e clique em Excluir.

   3. Aguarde até o ambiente ser excluído.

2. Exclua o bucket do ambiente. A exclusão do ambiente do Cloud Composer não exclui o bucket.

   1. No console do Google Cloud, acesse a página Armazenamento > Navegador.

      Acesse Armazenamento > Navegador

   2. Selecione o bucket do ambiente e clique em Excluir. Por exemplo, ele pode ter o nome us-central1-example-environ-c1616fe8-bucket.

3. Exclua o disco permanente da fila do Redis do seu ambiente. A exclusão do ambiente do Cloud Composer não remove o disco permanente.

   1. No console do Google Cloud, acesse a página Compute Engine > Discos.

      Acessar "Discos"

   2. Selecione o disco permanente da fila do Redis do ambiente e clique em Excluir.

      Por exemplo, ele pode ter o nome gke-us-central1-exampl-pvc-b12055b6-c92c-43ff-9de9-10f2cc6fc0ee. Os discos do Cloud Composer 1 sempre têm a Standard persistent disk tipo e o tamanho de 2 GB.

A seguir

• Como adicionar e atualizar DAGs
• Como escrever DAGs
• Como modificar as configurações do Airflow
• Como acessar a IU do Airflow