Fluxo de trabalho usando o Airflow Gerenciado

Objetivo:

1. Crie um modelo de fluxo de trabalho do Serviço Gerenciado para Apache Spark que execute um job do Spark PI
2. Crie um DAG do Apache Airflow que o Airflow Gerenciado usará para iniciar o fluxo de trabalho em um horário específico.

Neste documento, você vai usar os seguintes componentes faturáveis do Google Cloud:

• Managed Service for Apache Spark
• Compute Engine
• Managed Airflow

Para gerar uma estimativa de custo baseada na projeção de uso deste tutorial, use a calculadora de preços.

Novos usuários do Google Cloud podem estar qualificados para um teste sem custo financeiro.

Antes de começar

Criar o projeto

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

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.

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.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, and Managed Airflow APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Instale a CLI do Google Cloud.

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

Para inicializar a gcloud CLI, execute o seguinte comando:

gcloud init

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.

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.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, and Managed Airflow APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Instale a CLI do Google Cloud.

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

Para inicializar a gcloud CLI, execute o seguinte comando:

gcloud init

Criar um modelo de fluxo de trabalho do Serviço Gerenciado para Apache Spark

Copie e execute os comandos a seguir em uma janela de terminal local ou no Cloud Shell para criar e definir um modelo de fluxo de trabalho.

1. Crie o modelo de fluxo de trabalho sparkpi.

   gcloud dataproc workflow-templates create sparkpi \
       --region=us-central1
         

2. Adicione o job do Spark ao modelo de fluxo de trabalho sparkpi. A sinalização step-id "compute" identifica o job do SparkPi.

   gcloud dataproc workflow-templates add-job spark \
       --workflow-template=sparkpi \
       --step-id=compute \
       --class=org.apache.spark.examples.SparkPi \
       --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
       --region=us-central1 \
       -- 1000
         

3. Use um cluster gerenciado de nó único para executar o fluxo de trabalho. O Serviço Gerenciado para Apache Spark vai criar o cluster, executar o fluxo de trabalho nele e excluir o cluster quando o fluxo de trabalho for concluído.

   gcloud dataproc workflow-templates set-managed-cluster sparkpi \
       --cluster-name=sparkpi \
       --single-node \
       --region=us-central1
         

4. Confirme a criação do modelo de fluxo de trabalho.

   Console

   Clique no nome do sparkpi na página Fluxos de trabalho do Serviço gerenciado para Apache Spark no console Google Cloud para abrir a página Detalhes do modelo de fluxo de trabalho. Clique no nome do seu modelo de fluxo de trabalho para confirmar os atributos do modelo sparkpi.

   

   Comando gcloud

   Execute este comando:

   gcloud dataproc workflow-templates describe sparkpi --region=us-central1
       

Criar e fazer upload de um DAG para o Cloud Storage

1. Crie ou use um ambiente gerenciado do Airflow.
2. Definir variáveis de ambiente.

   IU do Airflow

   1. Na barra de ferramentas, clique em Administrador > Variáveis.
   2. Clique em Criar.
   3. Digite as seguintes informações:
      • Chave: project_id
      • Val: PROJECT_ID: o ID do projeto do Google Cloud
   4. Clique em Salvar.

   Comando gcloud

   Digite os seguintes comandos:

   • ENVIRONMENT é o nome do ambiente do Airflow gerenciado.
   • LOCATION é a região em que o ambiente do Airflow Gerenciado está localizado.
   • PROJECT_ID é o ID do projeto que contém o ambiente do Airflow Gerenciado.

       gcloud composer environments run ENVIRONMENT --location LOCATION variables set -- project_id PROJECT_ID
       

3. Copie o código do DAG a seguir localmente em um arquivo chamado "composer-dataproc-dag.py", que usa o DataprocInstantiateWorkflowTemplateOperator.

   Airflow 2

   
   """Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
   Spark Pi Job.
   
   This DAG relies on an Airflow variable
   https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
   * project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
   """
   
   import datetime
   
   from airflow import models
   from airflow.providers.google.cloud.operators.dataproc import (
       DataprocInstantiateWorkflowTemplateOperator,
   )
   from airflow.utils.dates import days_ago
   
   project_id = "{{var.value.project_id}}"
   
   
   default_args = {
       # Tell airflow to start one day ago, so that it runs as soon as you upload it
       "start_date": days_ago(1),
       "project_id": project_id,
   }
   
   # Define a DAG (directed acyclic graph) of tasks.
   # Any task you create within the context manager is automatically added to the
   # DAG object.
   with models.DAG(
       # The id you will see in the DAG airflow page
       "dataproc_workflow_dag",
       default_args=default_args,
       # The interval with which to schedule the DAG
       schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
   ) as dag:
       start_template_job = DataprocInstantiateWorkflowTemplateOperator(
           # The task id of your job
           task_id="dataproc_workflow_dag",
           # The template id of your workflow
           template_id="sparkpi",
           project_id=project_id,
           # The region for the template
           region="us-central1",
       )
   

   Airflow 1

   
   """Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
   Spark Pi Job.
   
   This DAG relies on an Airflow variable
   https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
   * project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
   """
   
   import datetime
   
   from airflow import models
   from airflow.contrib.operators import dataproc_operator
   from airflow.utils.dates import days_ago
   
   project_id = "{{var.value.project_id}}"
   
   
   default_args = {
       # Tell airflow to start one day ago, so that it runs as soon as you upload it
       "start_date": days_ago(1),
       "project_id": project_id,
   }
   
   # Define a DAG (directed acyclic graph) of tasks.
   # Any task you create within the context manager is automatically added to the
   # DAG object.
   with models.DAG(
       # The id you will see in the DAG airflow page
       "dataproc_workflow_dag",
       default_args=default_args,
       # The interval with which to schedule the DAG
       schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
   ) as dag:
       start_template_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
           # The task id of your job
           task_id="dataproc_workflow_dag",
           # The template id of your workflow
           template_id="sparkpi",
           project_id=project_id,
           # The region for the template
           # For more info on regions where Dataflow is available see:
           # https://cloud.google.com/dataflow/docs/resources/locations
           region="us-central1",
       )
   

4. Faça o upload do DAG para a pasta do ambiente no Cloud Storage. Depois que o upload for concluído, clique no link Pasta de DAGs na página do ambiente Airflow Gerenciado.
   

Ver o status de uma tarefa

IU do Airflow

1. Abra a interface da Web do Airflow.
2. Na página DAGs, clique no nome do DAG, por exemplo, dataproc_workflow_dag.
3. Na página "Detalhes dos DAGs", clique em Visualizar gráfico.
4. Verificar status:
   • Falha: a tarefa tem uma caixa vermelha ao redor. Você também pode manter o ponteiro do mouse sobre a tarefa e verificar se há a mensagem Estado: Falha.
   • Bem-sucedida: a tarefa tem uma caixa verde ao redor. Você também pode manter o ponteiro do mouse sobre a tarefa e verificar se há a mensagem Estado: sucesso.

Console

Clique na guia "Fluxos de trabalho" para ver o status do fluxo de trabalho.

Comando gcloud

gcloud dataproc operations list \
    --region=us-central1 \
    --filter="labels.goog-dataproc-workflow-template-id=sparkpi"
    

Limpar

Para evitar cobranças na sua conta do Google Cloud , exclua os recursos usados neste tutorial:

1. Exclua o ambiente do Airflow gerenciado.

2. Exclua o modelo de fluxo de trabalho.

A seguir

• Consulte Visão geral dos modelos de fluxo de trabalho do Serviço Gerenciado para Apache Spark.
• Consulte Soluções de programação de fluxo de trabalho.