O Cloud Composer 1 está no modo pós-manutenção. O Google não lança mais atualizações para o Cloud Composer 1, incluindo novas versões do Airflow, correções de bugs e atualizações de segurança. Recomendamos planejar a migração para o Cloud Composer 2.

Esta página foi traduzida pela API Cloud Translation.

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

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Neste tutorial, mostramos como usar o Cloud Composer para criar um DAG (gráfico acíclico direcionado) do Apache Airflow que executa um job de contagem de palavras do Apache Hadoop em um cluster do Dataproc.

Objetivos

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

Custos

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

Cloud Composer
Dataproc
Cloud Storage

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 uma avaliação gratuita.

Antes de começar

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

Enable the Dataproc, Cloud Storage APIs.
Enable the APIs
gcloud

Enable the Dataproc, Cloud Storage APIs:
```
gcloud services enable dataproc.googleapis.com storage-component.googleapis.com
```
No projeto, crie um bucket do Cloud Storage de qualquer classe e região de armazenamento para armazenar os resultados do job de contagem de palavras do Hadoop.

Observação: para requisitos de residência de dados ou benefícios de desempenho, crie o bucket de armazenamento na mesma região em que você planeja criar seu ambiente.
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 use a variável no DAG de exemplo mais adiante neste tutorial.
Crie um ambiente do Cloud Composer 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.
Observe a região em que você criou o ambiente, por exemplo, us-central. Você definirá uma variável do Airflow para essa região e usá-lo no DAG de exemplo para executar um cluster do Dataproc na mesma região.

definir variáveis do Airflow

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

Variável do Airflow	Valor
`gcp_project`	O ID do projeto que você está usando neste tutorial, como `example-project`.
`gcs_bucket`	O bucket do URI do Cloud Storage que você criou para este tutorial, como `gs://example-bucket.`.
`gce_region`	A região em que você criou o ambiente, como `us-central1`. Essa é a região em que o cluster do Dataproc será criado.

Conferir o fluxo de trabalho de exemplo

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 seguintes três operadores do Airflow:

DataprocClusterCreateOperator: cria um cluster do Dataproc.
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 recorrentes 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 sequencialmente.

# 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 a cada dia. Como o start_date transmitido para default_dag_args está definido como yesterday, o Cloud Composer programa o fluxo de trabalho para iniciar imediatamente após o upload do DAG 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:

Faça upload do DAG para o bucket do ambiente

O Cloud Composer armazena DAGs na pasta /dags da sua bucket do seu ambiente de execução.

Siga estas etapas para fazer o upload do DAG:

Na máquina local, salve hadoop_tutorial.py.
No console do Google Cloud, acesse a página Ambientes.

Acessar "Ambientes"
Na lista de ambientes, na coluna pasta de DAGs de cada ambiente, clique no link DAGs.
Clique em Enviar arquivos.
Selecione hadoop_tutorial.py na sua máquina local e clique em Abrir.

O Cloud Composer adiciona o DAG ao Airflow e agenda o DAG automaticamente. As alterações no DAG levam de 3 a 5 minutos.

Analisar execuções de DAG

Conferir o status da tarefa

Quando você faz o upload do arquivo DAG para a pasta dags/ no Cloud Storage, o Cloud Composer analisa o arquivo. Quando concluído, o nome do fluxo de trabalho aparece na listagem de DAGs e é colocado em fila para ser executado imediatamente.

Para ver o status da tarefa, acesse a interface da Web do Airflow e clique em DAGs na barra de ferramentas.
Para abrir a página de detalhes do DAG, clique em composer_hadoop_tutorial. Isso inclui uma representação gráfica das tarefas do fluxo de trabalho e dependências.
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:

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

Ver resultados de tarefas

Também é possível verificar o status e os resultados do composer_hadoop_tutorial. do BigQuery acessando as seguintes páginas do console do Google Cloud:

Clusters do Dataproc: para monitorar a criação de clusters e exclusão. O cluster criado pelo fluxo de trabalho é temporário: ele existe apenas durante o fluxo de trabalho e é excluído como parte do a última tarefa do fluxo de trabalho.

Acesse Clusters do Dataproc
Jobs do Dataproc: para visualizar 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 Dataproc
Navegador do Cloud Storage: para ver os resultados da contagem de palavras no a pasta wordcount no bucket do Cloud Storage que você criou para este tutorial.

Acesse Navegador do Cloud Storage

Limpeza

Exclua os recursos usados neste tutorial:

Exclua o ambiente do Cloud Composer, incluindo: excluir manualmente o bucket do ambiente.
Exclua o bucket do Cloud Storage que armazena os resultados do job de contagem de palavras do Hadoop.