Como executar um DAG do Apache Airflow no Cloud Composer 1

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Neste guia de início rápido, mostramos como criar um ambiente do Cloud Composer e execute um DAG do Apache Airflow no Cloud Composer 1.

Antes de começar

  1. Faça login na sua conta do Google Cloud. Se você começou a usar o Google Cloud agora, crie uma conta para avaliar o desempenho de 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. No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.

    Acessar o seletor de projetos

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

  4. No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.

    Acessar o seletor de projetos

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

  6. Ative a API Cloud Composer.

    Ative 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 como conceder papéis, consulte Gerenciar acesso.

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

Criar um ambiente

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

    Acessar "Criar ambiente"

  2. No campo Nome, use example-environment.

  3. Na lista suspensa Local, selecione uma região para o ambiente do Cloud Composer. 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 até que o ambiente seja criado.

  6. Quando terminar, uma marca de seleção verde vai aparecer ao lado do nome do ambiente.

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. Este 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

Todo ambiente do Cloud Composer tem em um bucket associado a ela. Somente as programações do Airflow no Cloud Composer DAGs que estão localizados na pasta /dags neste bucket.

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

  1. No console do Google Cloud, 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ê enviou. Pode levar alguns minutos para que 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 que você forneceu no arquivo DAG.

Verifique se o DAG foi processado sem erros e está disponível no Airflow na 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 à IU do Airflow, que é uma interface nativa da plataforma interface gráfica do usuário.

  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. No console do Google Cloud, 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. Verificar se o DAG composer_quickstart está presente na lista de DAGs.

    A lista de DAGs exibe o DAG composer_quickstart com informações adicionais, como estado e horário
    Figura 1. A lista de DAGs exibe DAG composer_quickstart (clique para ampliar)
    .

Mais detalhes da execução do DAG

Uma única execução de um DAG é chamada de execução de DAG. Airflow imediatamente executa um DAG para o DAG de exemplo porque a data de início no arquivo DAG é definido como ontem. Dessa forma, o Airflow alcança os DAGs especificados cronograma.

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

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

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

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

  3. Os detalhes da execução do DAG são exibidos com as informações individuais tarefas do DAG de exemplo.

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

  4. A seção Registros para 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 é a Saída e as outras listas 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 estas etapas.

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 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, esse disco pode ser nomeado 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