Executar um DAG do Apache Airflow no Airflow gerenciado (geração 2)

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

Este guia de início rápido mostra como criar um ambiente do Serviço gerenciado para Apache Airflow e executar um DAG do Apache Airflow no Airflow gerenciado (geração 2).

Antes de começar

  1. Faça login na sua Google Cloud conta do. Se você começou a usar o Google Cloud, 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.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  5. Verify that billing is enabled for your Google Cloud project.

  6. Ative a API do Airflow Gerenciado.

    Funções necessárias para ativar APIs

    Para ativar as APIs, é necessário ter o papel do IAM de administrador de uso do serviço (roles/serviceusage.serviceUsageAdmin), que contém a permissão serviceusage.services.enable. Saiba como conceder papéis.

    Ativar a API

  7. 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 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 personalizados papéis ou outros predefinidos papéis.

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. O 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 o ambiente:

  1. Crie uma nova conta de serviço, conforme descrito em a 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

  1. No Google Cloud console, acesse a página Criar ambiente.

    Acessar "Criar ambiente"

  1. Se este for o primeiro ambiente no projeto, uma seção Conceder permissões necessárias à conta de serviço do Cloud Composer será exibida.

    Adicione a conta do agente de serviço do Airflow Gerenciado como um novo principal na conta de serviço do ambiente e conceda a ela o papel Extensão do agente de serviço da API Cloud Composer v2.

    Confirme se você está usando a conta de serviço pretendida para o ambiente e clique em Conceder.

  2. No campo Nome, use example-environment.

  3. Na lista suspensa Local, selecione uma região para o ambiente do Airflow Gerenciado. Este guia usa a região us-central1.

  4. Para outras opções de configuração do ambiente, use os padrões fornecidos.

  5. Clique em Criar e aguarde a criação do ambiente.

  6. Quando isso acontecer, uma marca de seleção verde será exibida ao lado do nome do ambiente.

Criar um arquivo DAG

Um Airflow DAG é 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 a ele. O Airflow no Airflow Gerenciado programa apenas DAGs 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:

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

    Acessar "Ambientes"

  2. Na lista de ambientes, clique no nome do ambiente, example-environment. A página Detalhes do ambiente é aberta.

  3. Clique em Abrir pasta de DAGs. A página Detalhes do bucket é aberta.

  4. Clique em Fazer upload de arquivos e selecione sua cópia de quickstart.py.

  5. Para fazer upload do arquivo, clique em Abrir.

Visualizar o DAG

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

  1. Analisa o arquivo DAG que você fez upload. 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 visualizando-o na interface do DAG. A interface do DAG é a interface do Airflow Gerenciado para visualizar informações do DAG no Google Cloud console. 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ê fez upload anteriormente e conclua a primeira execução do DAG (explicada mais adiante).

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

    Acessar "Ambientes"

  3. Na lista de ambientes, clique no nome do ambiente, example-environment. A página Detalhes do ambiente é aberta.

  4. Acesse a guia DAGs.

  5. Verifique se o DAG composer_quickstart está presente na lista de DAGs.

    A lista de DAGs mostra o DAG composer_quickstart com informações adicionais, como estado e programação.
    Figura 1. A lista de DAGs mostra o DAG composer_quickstart (clique para ampliar)

Visualizar 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 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).

  1. Na guia DAGs, clique em composer_quickstart. A guia Execuções do DAG é aberta.

  2. Na lista de execuções de DAG, clique na primeira entrada.

    A lista de execuções de DAG mostra a execução recente (data de execução e status).
    Figura 2. A lista de execuções de DAG para o DAG composer_quickstart (clique para ampliar)

  3. Os detalhes da execução do DAG são mostrados, detalhando as informações sobre tarefas individuais do DAG de exemplo.

    A lista de tarefas com uma entrada print_dag_run_conf, o horário de início, o horário de término e a duração
    Figura 3. A lista de tarefas que foram executadas em na execução do DAG (clique para ampliar)

  4. A seção Registros da execução do DAG lista os registros de todas as tarefas na execução do DAG. É possível conferir a saída do comando echo nos registros.

    Entradas de registro da tarefa. Uma delas é "Output", e a outra lista um identificador.
    Figura 4. Registros da tarefa print_dag_run_conf (clique para ampliar)

Limpar

Para evitar cobranças na conta do Google Cloud pelos recursos usados nesta página, siga as etapas abaixo.

Exclua os recursos usados neste tutorial:

  1. Exclua o ambiente do Airflow Gerenciado:

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

      Acesse Armazenamento > Navegador

    2. Selecione o bucket do ambiente e clique em Excluir. Por exemplo, esse bucket pode ser chamado de us-central1-example-environ-c1616fe8-bucket.

A seguir