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

Antes de começar

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.

Install the Google Cloud CLI.

Se você estiver usando um provedor de identidade externo (IdP), primeiro faça login na CLI gcloud com sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte comando:

gcloud init

Create or select a Google Cloud 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.

Verify that billing is enabled for your Google Cloud project.

Install the Google Cloud CLI.

Se você estiver usando um provedor de identidade externo (IdP), primeiro faça login na CLI gcloud com sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte comando:

gcloud init

Create or select a Google Cloud 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.

Verify that billing is enabled for your Google Cloud project.

2. Enable the Cloud Composer API:

   gcloud services enable composer.googleapis.com

3. 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 ver, criar e gerenciar o ambiente do Cloud Composer:
     • Administrador de objetos do armazenamento e do ambiente (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 uma conta de serviço de um ambiente

Ao criar um ambiente, você especifica uma conta de serviço. Essa conta de serviço é chamada de conta de serviço do ambiente. Seu ambiente usa essa conta de serviço para realizar a maioria das operações.

A conta de serviço do ambiente não é uma conta de usuário. Uma conta de serviço é um tipo especial de conta usada por um aplicativo ou uma instância de máquina virtual (VM), não uma pessoa.

Para criar uma conta de serviço para seu ambiente:

1. Crie uma conta de serviço, conforme descrito na documentação do Identity and Access Management.

2. Conceda um papel a ele, conforme descrito na documentação do Identity and Access Management. O papel necessário é Worker do Composer (composer.worker).

Criar um ambiente

Se este for o primeiro ambiente no seu projeto, adicione a conta do agente de serviço do Cloud Composer como um novo principal na conta de serviço do ambiente e conceda a ela o papel roles/composer.ServiceAgentV2Ext.

Por padrão, seu ambiente usa a conta de serviço padrão do Compute Engine, e o exemplo a seguir mostra como adicionar essa permissão a ela.

# Get current project's project number
PROJECT_NUMBER=$(gcloud projects list \
  --filter="$(gcloud config get-value project)" \
  --format="value(PROJECT_NUMBER)" \
  --limit=1)

# Add the Cloud Composer v2 API Service Agent Extension role
gcloud iam service-accounts add-iam-policy-binding \
    $PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --member serviceAccount:service-$PROJECT_NUMBER@cloudcomposer-accounts.iam.gserviceaccount.com \
    --role roles/composer.ServiceAgentV2Ext

Crie um ambiente chamado example-environment na região us-central1 com a versão mais recente do Cloud Composer 2.

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-2.13.9-airflow-2.10.5

Criar um arquivo DAG

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

Este guia usa um exemplo de DAG do Airflow definido no arquivo quickstart.py. O código Python neste 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 no bucket do ambiente.

Cada ambiente do Cloud Composer tem um bucket do Cloud Storage associado. O Airflow no Cloud Composer agenda somente DAGs que estão na pasta /dags nesse bucket.

Para agendar 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

Ver o DAG

Depois de fazer o 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 foi processado sem erros e está disponível no Airflow. Para isso, confira na interface do usuário do DAG. A interface do usuário do DAG é a interface do Cloud Composer para visualizar informações do DAG no console 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 processe o arquivo DAG que você enviou antes e conclua a primeira execução do DAG (explicada mais adiante).

2. Execute o comando a seguir na Google Cloud CLI. Esse comando executa o dags list comando da CLI do Airflow que lista os DAGs no seu ambiente.

   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
   

Ver 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 DAG de exemplo porque a data de início no arquivo DAG está definida como "ontem". Dessa forma, o Airflow acompanha a programação do DAG especificado.

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

Execute o comando a seguir na Google Cloud CLI. Esse comando lista as execuções do 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

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

Execute o comando a seguir na Google Cloud CLI. Esse comando lê os registros do Cloud Logging para uma execução específica do DAG composer_quickstart. O argumento --format formata a saída para que apenas o texto da mensagem de registro seja 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 conta do Google Cloud pelos recursos usados nesta página, exclua o projeto do Google Cloud com os recursos.

Exclua os recursos usados neste tutorial:

1. Exclua o ambiente do Cloud Composer:

   1. No console 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 Google Cloud , acesse a página Storage > 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.

A seguir

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