Executar um DAG do Apache Airflow no Airflow Gerenciado (3ª geração) (Google Cloud CLI)

Executar um DAG do Apache Airflow no Airflow Gerenciado (3ª geração) (Google Cloud CLI)

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

Neste guia de início rápido, mostramos como criar um ambiente do Serviço Gerenciado para Apache Airflow e executar um DAG do Apache Airflow no Airflow Gerenciado (3ª geração).

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

• Se você quiser usar o console do Google Cloud , consulte Executar um DAG do Apache Airflow no Serviço Gerenciado para Apache Airflow.

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

Antes de começar

Faça login na sua conta do Google Cloud . Se você começou a usar o Google Cloudagora, crie uma conta para avaliar o desempenho dos nossos produtos em situações reais. Clientes novos também recebem US $300 em créditos para executar, testar e implantar cargas de trabalho sem custos financeiros.

Instale a CLI do Google Cloud.

Ao usar um provedor de identidade (IdP) externo, primeiro faça login na CLI gcloud com sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte comando:

gcloud init

Crie ou selecione um projeto do Google Cloud .

Papéis necessários para selecionar ou criar um projeto

• Selecionar um projeto: para selecionar um projeto, não é necessário ter um papel do IAM específico. É possível selecionar qualquer projeto em que você tenha um papel.
• Criar um projeto: para criar um projeto, é preciso ter o papel de Criador de projetos (roles/resourcemanager.projectCreator), que contém a permissão resourcemanager.projects.create. Saiba como conceder papéis.

Observação: se você não pretende manter os recursos criados neste procedimento, crie um projeto novo em vez de selecionar um que já existe. Depois de concluir essas etapas, é possível excluir o projeto. Para fazer isso, basta remover todos os recursos associados a ele.

• Crie um projeto do Google Cloud :

  gcloud projects create PROJECT_ID

  Substitua PROJECT_ID por um nome para o projeto do Google Cloud que você está criando.

• Selecione o projeto do Google Cloud que você criou:

  gcloud config set project PROJECT_ID

  Substitua PROJECT_ID pelo nome do projeto do Google Cloud .

Verifique se o faturamento está ativado para o projeto do Google Cloud .

Instale a CLI do Google Cloud.

Ao usar um provedor de identidade (IdP) externo, primeiro faça login na CLI gcloud com sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte comando:

gcloud init

Crie ou selecione um projeto do Google Cloud .

Papéis necessários para selecionar ou criar um projeto

• Selecionar um projeto: para selecionar um projeto, não é necessário ter um papel do IAM específico. É possível selecionar qualquer projeto em que você tenha um papel.
• Criar um projeto: para criar um projeto, é preciso ter o papel de Criador de projetos (roles/resourcemanager.projectCreator), que contém a permissão resourcemanager.projects.create. Saiba como conceder papéis.

Observação: se você não pretende manter os recursos criados neste procedimento, crie um projeto novo em vez de selecionar um que já existe. Depois de concluir essas etapas, é possível excluir o projeto. Para fazer isso, basta remover todos os recursos associados a ele.

• Crie um projeto do Google Cloud :

  gcloud projects create PROJECT_ID

  Substitua PROJECT_ID por um nome para o projeto do Google Cloud que você está criando.

• Selecione o projeto do Google Cloud que você criou:

  gcloud config set project PROJECT_ID

  Substitua PROJECT_ID pelo nome do projeto do Google Cloud .

Verifique se o faturamento está ativado para o projeto do Google Cloud .

1. Ative a API do Airflow Gerenciado, se ainda não estiver ativada:

   Funções necessárias para ativar APIs

   Para ativar APIs, você precisa da permissão serviceusage.services.enable. Se você criou o projeto, provavelmente já tem essa permissão pelo papel de proprietário (roles/owner). Caso contrário, é possível conseguir essa permissão pelo papel de administrador do Service Usage (roles/serviceusage.serviceUsageAdmin). Saiba como conceder papéis.

   gcloud services enable composer.googleapis.com

2. 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 atribuir papéis e permissões do IAM: Administrador do IAM do projeto (roles/resourcemanager.projectIamAdmin)
   • Para criar uma conta de serviço para o ambiente do Airflow Gerenciado: Criar contas de serviço (roles/iam.serviceAccountCreator)
   • Para visualizar, criar e gerenciar o ambiente do Airflow Gerenciado:
     • Administrador de objetos de armazenamento e ambiente (roles/composer.environmentAndStorageObjectAdmin)
     • Usuário da conta de serviço (roles/iam.serviceAccountUser)
   • Para ver 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 usando papéis personalizados ou outros papéis predefinidos.

Criar uma conta de serviço do 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 seu 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 nova conta de serviço, conforme descrito na documentação do Identity and Access Management.

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

Criar um ambiente

Crie um novo ambiente chamado example-environment na região us-central1, com a versão mais recente do Airflow Gerenciado (3ª geração).

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-3-airflow-2.11.1-build.21 \
    --service-account ENVIRONMENT_SERVICE_ACCOUNT

Substitua ENVIRONMENT_SERVICE_ACCOUNT pela conta de serviço do seu ambiente que você criou anteriormente.

Observação : o tempo aproximado para criar um ambiente é de 25 minutos.

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 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 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 }}"
    )

Fazer upload do arquivo DAG para o bucket do ambiente

Cada ambiente do Airflow gerenciado tem um bucket do Cloud Storage associado. O Airflow no Airflow Gerenciado programa apenas DAGs que estão localizados na pasta /dags nesse 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 seguinte comando 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 que você faz upload do arquivo DAG, o Airflow faz o seguinte:

1. Analisa o arquivo DAG que você enviou. Pode levar alguns minutos para o DAG ficar 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 se está disponível no Airflow visualizando-o na interface do DAG. A interface do DAG é a interface do Airflow gerenciado para visualizar informações do DAG no console do Google Cloud . O Airflow Gerenciado 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 anteriormente 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
   

   Observação: o DAG airflow_monitoring é usado pelo Airflow Gerenciado para monitorar a integridade do ambiente.

Ver detalhes da execução do DAG

Uma única execução de um DAG é chamada de execução do DAG. O Airflow executa imediatamente uma execução do DAG de exemplo porque a data de início no arquivo DAG está definida como ontem. Dessa forma, o Airflow se atualiza com 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. Este comando lista as execuções do 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: use o comando tasks states-for-dag-run para ver o status de tarefas individuais.

A CLI do Airflow não fornece um comando para visualizar registros de tarefas. É possível usar outros métodos para visualizar registros de tarefas do Airflow: interface de DAG do Airflow gerenciado, interface do Airflow ou Cloud Logging. Neste guia, mostramos como consultar o Cloud Logging para encontrar registros de uma execução de DAG específica.

Execute o comando a seguir na Google Cloud CLI. Esse comando lê 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 pelo 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 e os recursos.

Exclua os recursos usados neste tutorial:

1. Exclua o ambiente do Airflow Gerenciado:

   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 Airflow Gerenciado 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.

A seguir

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