Executar um job de contagem de palavras do Hadoop em um cluster do Cloud Dataproc

Neste tutorial, mostramos como usar o Serviço gerenciado para Apache Airflow para criar um DAG (gráfico acíclico dirigido) do Apache Airflow que executa um job de contagem de palavras do Apache Hadoop em um cluster do Serviço gerenciado para Apache Spark.

Objetivos

1. Acesse seu ambiente do Airflow gerenciado e use a interface do Airflow.
2. Crie e visualize variáveis de ambiente do Airflow.
3. Criar e executar um DAG que inclua as seguintes tarefas:
   1. Cria um cluster do Serviço Gerenciado para Apache Spark.
   2. Executa um job de contagem de palavras do Apache Hadoop (em inglês) no cluster.
   3. Gera os resultados da contagem de palavras em um bucket do Cloud Storage.
   4. Exclui o cluster.

Custos

Neste documento, você vai usar os seguintes componentes faturáveis do Cloud de Confiance by S3NS:

• Managed Airflow
• Managed Service for Apache Spark
• Cloud Storage

Antes de começar

• Verifique se as seguintes APIs estão ativadas no seu projeto:

  Console

  Ative o Serviço gerenciado para Apache Spark e as APIs Cloud Storage, se ainda não estiverem ativados.

  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 com o papel de Proprietário (roles/owner). Caso contrário, é possível receber essa permissão com o papel de Administrador do Service Usage (roles/serviceusage.serviceUsageAdmin). Saiba como conceder papéis.

  Ativar as APIs

  gcloud

  Ative o Serviço Gerenciado para Apache Spark e as APIs Cloud Storage, se ainda não estiverem ativados:

  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 com o papel de Proprietário (roles/owner). Caso contrário, é possível receber essa permissão com o papel de Administrador do Service Usage (roles/serviceusage.serviceUsageAdmin). Saiba como conceder papéis.

  gcloud services enable dataproc.googleapis.com storage-component.googleapis.com

• No seu projeto, crie um bucket do Cloud Storage de qualquer classe de armazenamento e região para armazenar os resultados do job de contagem de palavras do Hadoop.

• Anote o caminho do bucket que você criou, por exemplo, gs://example-bucket. Você vai definir uma variável do Airflow para esse caminho e usar a variável no exemplo de DAG mais adiante neste tutorial.

• Crie um ambiente do Airflow Gerenciado com parâmetros padrão. Aguarde até que a criação do ambiente seja concluída. Quando terminar, a marca de seleção verde vai aparecer à esquerda do nome do ambiente.

• Anote a região em que você criou o ambiente, por exemplo, us-central. Você vai definir uma variável do Airflow para essa região e usá-la no DAG de exemplo para executar um cluster do Serviço Gerenciado para Apache Spark na mesma região.

Definir variáveis do Airflow

Defina as variáveis do Airflow para usar mais tarde no exemplo de DAG. Por exemplo, é possível definir variáveis do Airflow na interface do Airflow.

Ver o exemplo de fluxo de trabalho

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. O código mostrado em hadoop_tutorial.py é o código do fluxo de trabalho.

"""Example Airflow DAG that creates a Cloud Dataproc cluster, runs the Hadoop
wordcount example, and deletes the cluster.

This DAG relies on three Airflow variables
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
* gce_region - Google Compute Engine region where Cloud Dataproc cluster should be
  created.
* gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
  See https://cloud.google.com/storage/docs/creating-buckets for creating a
  bucket.
"""

import datetime
import os

from airflow import models
from airflow.providers.google.cloud.operators import dataproc
from airflow.utils import trigger_rule

# Output file for Cloud Dataproc job.
# 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
output_file = (
    os.path.join(
        "{{ var.value.gcs_bucket }}",
        "wordcount",
        datetime.datetime.now().strftime("%Y%m%d-%H%M%S"),
    )
    + os.sep
)
# Path to Hadoop wordcount example available on every Dataproc cluster.
WORDCOUNT_JAR = "file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar"
# Arguments to pass to Cloud Dataproc job.
input_file = "gs://pub/shakespeare/rose.txt"
wordcount_args = ["wordcount", input_file, output_file]

HADOOP_JOB = {
    "reference": {"project_id": "{{ var.value.gcp_project }}"},
    "placement": {"cluster_name": "composer-hadoop-tutorial-cluster-{{ ds_nodash }}"},
    "hadoop_job": {
        "main_jar_file_uri": WORDCOUNT_JAR,
        "args": wordcount_args,
    },
}

CLUSTER_CONFIG = {
    "master_config": {"num_instances": 1, "machine_type_uri": "n1-standard-2"},
    "worker_config": {"num_instances": 2, "machine_type_uri": "n1-standard-2"},
}

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()
)

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    "start_date": yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    "email_on_failure": False,
    "email_on_retry": False,
    # If a task fails, retry it once after waiting at least 5 minutes
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "project_id": "{{ var.value.gcp_project }}",
    "region": "{{ var.value.gce_region }}",
}


with models.DAG(
    "composer_hadoop_tutorial",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

    # Create a Cloud Dataproc cluster.
    create_dataproc_cluster = dataproc.DataprocCreateClusterOperator(
        task_id="create_dataproc_cluster",
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        cluster_config=CLUSTER_CONFIG,
        region="{{ var.value.gce_region }}",
    )

    # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
    # master node.
    run_dataproc_hadoop = dataproc.DataprocSubmitJobOperator(
        task_id="run_dataproc_hadoop", job=HADOOP_JOB
    )

    # Delete Cloud Dataproc cluster.
    delete_dataproc_cluster = dataproc.DataprocDeleteClusterOperator(
        task_id="delete_dataproc_cluster",
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        region="{{ var.value.gce_region }}",
        # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
        # even if the Dataproc job fails.
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE,
    )

    # Define DAG dependencies.
    create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Operadores

Para orquestrar as três tarefas no fluxo de trabalho de exemplo, o DAG importa os três operadores do Airflow a seguir:

• DataprocClusterCreateOperator: cria um cluster do Serviço Gerenciado para Apache Spark.

• DataProcHadoopOperator: envia um job de contagem de palavras do Hadoop e grava os resultados em um bucket do Cloud Storage.

• DataprocClusterDeleteOperator: exclui o cluster para evitar cobranças contínuas do Compute Engine.

Dependências

Você organiza as tarefas que quer executar de uma forma que reflita os relacionamentos e as dependências delas. As tarefas neste DAG são executadas em sequência.

# Define DAG dependencies.
create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Programação

O nome do DAG é composer_hadoop_tutorial, e ele é executado uma vez por dia. Como o start_date transmitido para default_dag_args está definido como yesterday, o Airflow gerenciado programa o fluxo de trabalho para iniciar imediatamente após o DAG ser enviado para o bucket do ambiente.

with models.DAG(
    "composer_hadoop_tutorial",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:

Fazer upload do DAG para o bucket do ambiente

O Airflow Gerenciado armazena DAGs na pasta /dags no bucket do ambiente.

Siga estas etapas para fazer o upload do DAG:

1. Na máquina local, salve hadoop_tutorial.py.

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

   Acessar "Ambientes"

3. Na lista de ambientes, na coluna Pasta de DAGs do seu ambiente, clique no link DAGs.

4. Clique em Enviar arquivos.

5. Selecione hadoop_tutorial.py na sua máquina local e clique em Abrir.

O Airflow gerenciado adiciona e programa o DAG automaticamente. As alterações no DAG levam de 3 a 5 minutos.

Analisar execuções de DAG

Ver o status da tarefa

Quando você faz upload do arquivo DAG para a pasta dags/ no Cloud Storage, o Airflow Gerenciado analisa o arquivo. Quando concluído, o nome do fluxo de trabalho aparece na lista de DAGs e entra na fila para ser executado imediatamente.

1. Para ver o status da tarefa, acesse a interface da Web do Airflow e clique em DAGs na barra de ferramentas.

   

2. Para abrir a página de detalhes do DAG, clique em composer_hadoop_tutorial. Nela, você encontra uma representação gráfica das tarefas e das dependências do fluxo de trabalho.

   

3. Para ver o status de cada tarefa, clique em Visualização de gráfico e, em seguida, passe o mouse sobre a imagem de cada tarefa.

   

Enfileirar o fluxo de trabalho novamente

Para executar o fluxo de trabalho novamente a partir da Visualização de gráfico:

1. Na visualização de gráfico da IU do Airflow, clique no gráfico create_dataproc_cluster.
2. Para redefinir as três tarefas, clique em Limpar e em OK para confirmar.
3. Clique em create_dataproc_cluster novamente na visualização de gráfico.
4. Para enfileirar o fluxo de trabalho novamente, clique em Executar.

Conferir resultados da tarefa

Você também pode verificar o status e os resultados do fluxo de trabalho composer_hadoop_tutorial acessando as seguintes páginas do console Cloud de Confiance :

• Clusters do Serviço Gerenciado para Apache Spark: para monitorar a criação e exclusão de clusters. O cluster criado pelo fluxo de trabalho é efêmero: ele só existe durante a execução do fluxo e é excluído como parte da última tarefa.

  Acessar clusters do Serviço Gerenciado para Apache Spark

• Jobs do Serviço Gerenciado para Apache Spark: para ver ou monitorar o job de contagem de palavras do Apache Hadoop. Clique no ID do job para ver a saída do registro dele.

  Acessar jobs do Serviço Gerenciado para Apache Spark

• Navegador do Cloud Storage: para ver os resultados da contagem de palavras na pasta wordcount do bucket do Cloud Storage criado para este tutorial.

  Acesse Navegador do Cloud Storage

Limpeza

Exclua os recursos usados neste tutorial:

1. Exclua o ambiente do Airflow Gerenciado, incluindo a exclusão manual do bucket do ambiente.

2. Exclua o bucket do Cloud Storage que armazena os resultados do job de contagem de palavras do Hadoop.