Executar cargas de trabalho sem servidor do Dataproc com o Cloud Composer

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Observação :esta página ainda não foi revisada sobre o Cloud Composer 3 e mostra do Cloud Composer 2.

Nesta página, descrevemos como usar o Cloud Composer 2 para executar Cargas de trabalho Dataproc sem servidor no Google Cloud.

Os exemplos nas seções a seguir mostram como usar operadores para gerenciar cargas de trabalho em lote sem servidor do Dataproc. Use estes operadores em DAGs que criam, excluem, listam e extraem uma carga de trabalho em lote do Dataproc Serverless Spark:

• Criar DAGs para operadores que funcionam com cargas de trabalho em lote sem servidor do Dataproc:

  • DataprocCreateBatchOperator
  • DataprocDeleteBatchOperator
  • DataprocListBatchesOperator
  • DataprocGetBatchOperator

• Crie DAGs que usam contêineres personalizados e Metastore do Dataproc.

• Configure o servidor de histórico permanente para esses DAGs.

Antes de começar

1. Ative a API Dataproc:

   Console

   Enable the Dataproc API.

   Enable the API

   gcloud

   Enable the Dataproc API:

   gcloud services enable dataproc.googleapis.com

2. Selecione o local do seu arquivo de carga de trabalho em lote. É possível usar qualquer um seguintes opções:

   • Crie um bucket do Cloud Storage que armazene esse arquivo.
   • Use o bucket do ambiente. Como você não precisa sincronizar esse arquivo com o Airflow, crie uma subpasta separada fora das pastas /dags ou /data. Por exemplo, /batches.
   • Usar um bucket atual.

configurar arquivos e variáveis do Airflow

Nesta seção, demonstramos como configurar arquivos e variáveis do Airflow para este tutorial.

Fazer upload de um arquivo de carga de trabalho do Dataproc Serverless Spark ML para um bucket

A carga de trabalho neste tutorial executa um script pyspark:

1. Salve qualquer script pyspark em um arquivo local chamado spark-job.py. Por exemplo, é possível usar exemplo de script do pyspark.

2. Faça upload do arquivo no local selecionado. em Antes de começar.

definir variáveis do Airflow

Os exemplos nas seções a seguir usam variáveis do Airflow. Você define valores para essas variáveis no Airflow, e o código do DAG pode acessar esses valores.

Os exemplos neste tutorial usam as seguintes variáveis do Airflow. É possível defini-los conforme necessário, dependendo do exemplo usado.

Defina as seguintes variáveis do Airflow para usar no seu código DAG:

• project_id: ID do projeto.
• bucket_name: URI de um bucket em que o arquivo principal do Python da carga de trabalho (spark-job.py) está localizado. Você selecionou este local em Antes de começar.
• phs_cluster: nome do cluster do servidor de histórico permanente. Defina essa variável ao criar um servidor de histórico permanente.
• image_name: nome e tag da imagem do contêiner personalizado (image:tag). Você defina essa variável quando usar imagem de contêiner personalizada com DataprocCreateBatchOperator.
• metastore_cluster: nome do serviço do metastore do Dataproc. Defina essa variável ao usar o serviço do Dataproc Metastore com DataprocCreateBatchOperator.
• region_name: região em que o serviço Metastore do Dataproc está localizado. Defina essa variável ao usar o serviço do Dataproc Metastore com DataprocCreateBatchOperator.

Usar o console do Google Cloud e a interface do Airflow para definir cada variável do Airflow

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

   Acessar "Ambientes"

2. Na lista de ambientes, clique no link Airflow para sua de nuvem. A interface do Airflow é aberta.

3. Na interface do Airflow, selecione Admin > Variables.

4. Clique em Add a new record.

5. Especifique o nome da variável no campo Chave e defina o valor dela no campo Val.

6. Clique em Salvar.

Crie um servidor de histórico permanente

Use um Persistent History Server (PHS) para ver os arquivos de histórico do Spark do seu lote cargas de trabalho:

1. Crie um servidor de histórico permanente.
2. Verifique se você especificou o nome do cluster de PHS na variável Airflow phs_cluster.

DataprocCreateBatchOperator

O DAG a seguir inicia uma carga de trabalho do Dataproc Serverless Batch.

Para mais informações sobre argumentos DataprocCreateBatchOperator, consulte código-fonte do operador.

Para mais informações sobre os atributos que podem ser transmitidos no método batch parâmetro de DataprocCreateBatchOperator, consulte a descrição da classe Batch.

"""
Examples below show how to use operators for managing Dataproc Serverless batch workloads.
 You use these operators in DAGs that create, delete, list, and get a Dataproc Serverless Spark batch workload.
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id is the Google Cloud Project ID to use for the Cloud Dataproc Serverless.
* bucket_name is the URI of a bucket where the main python file of the workload (spark-job.py) is located.
* phs_cluster is the Persistent History Server cluster name.
* image_name is the name and tag of the custom container image (image:tag).
* metastore_cluster is the Dataproc Metastore service name.
* region_name is the region where the Dataproc Metastore service is located.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
    DataprocDeleteBatchOperator,
    DataprocGetBatchOperator,
    DataprocListBatchesOperator,
)
from airflow.utils.dates import days_ago

PROJECT_ID = "{{ var.value.project_id }}"
REGION = "{{ var.value.region_name}}"
BUCKET = "{{ var.value.bucket_name }}"
PHS_CLUSTER = "{{ var.value.phs_cluster }}"
METASTORE_CLUSTER = "{{var.value.metastore_cluster}}"
DOCKER_IMAGE = "{{var.value.image_name}}"

PYTHON_FILE_LOCATION = "gs://{{var.value.bucket_name }}/spark-job.py"
# for e.g.  "gs//my-bucket/spark-job.py"
# Start a single node Dataproc Cluster for viewing Persistent History of Spark jobs
PHS_CLUSTER_PATH = "projects/{{ var.value.project_id }}/regions/{{ var.value.region_name}}/clusters/{{ var.value.phs_cluster }}"
# for e.g. projects/my-project/regions/my-region/clusters/my-cluster"
SPARK_BIGQUERY_JAR_FILE = "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar"
# use this for those pyspark jobs that need a spark-bigquery connector
# https://cloud.google.com/dataproc/docs/tutorials/bigquery-connector-spark-example
# Start a Dataproc MetaStore Cluster
METASTORE_SERVICE_LOCATION = "projects/{{var.value.project_id}}/locations/{{var.value.region_name}}/services/{{var.value.metastore_cluster }}"
# for e.g. projects/my-project/locations/my-region/services/my-cluster
CUSTOM_CONTAINER = "us.gcr.io/{{var.value.project_id}}/{{ var.value.image_name}}"
# for e.g. "us.gcr.io/my-project/quickstart-image",

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,
    "region": REGION,
}
with models.DAG(
    "dataproc_batch_operators",  # The id you will see in the DAG airflow page
    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:
    create_batch = DataprocCreateBatchOperator(
        task_id="batch_create",
        batch={
            "pyspark_batch": {
                "main_python_file_uri": PYTHON_FILE_LOCATION,
                "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
            },
            "environment_config": {
                "peripherals_config": {
                    "spark_history_server_config": {
                        "dataproc_cluster": PHS_CLUSTER_PATH,
                    },
                },
            },
        },
        batch_id="batch-create-phs",
    )
    list_batches = DataprocListBatchesOperator(
        task_id="list-all-batches",
    )

    get_batch = DataprocGetBatchOperator(
        task_id="get_batch",
        batch_id="batch-create-phs",
    )
    delete_batch = DataprocDeleteBatchOperator(
        task_id="delete_batch",
        batch_id="batch-create-phs",
    )
    create_batch >> list_batches >> get_batch >> delete_batch

Usar a imagem de contêiner personalizada com o DataprocCreateBatchOperator

O exemplo a seguir mostra como usar uma imagem de contêiner personalizada para executar seu do Google Cloud. É possível usar um contêiner personalizado, por exemplo, para adicionar Python dependências não fornecidas pela imagem de contêiner padrão.

Observação: para executar DAGs que executam o Dataproc sem servidor para Spark é executado com contêineres personalizados do Docker, o ambiente do Cloud Composer precisa usar o Cloud Composer 2.0.6 e o Airflow 2.2.3 ou posterior mais recentes.

Para usar uma imagem de contêiner personalizada:

1. Crie uma imagem de contêiner personalizada e faça upload dela para o Container Registry.

2. Especifique a imagem na variável do Airflow image_name.

3. Use DataprocCreateBatchOperator com sua imagem personalizada:

create_batch_with_custom_container = DataprocCreateBatchOperator(
    task_id="dataproc_custom_container",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
        "runtime_config": {
            "container_image": CUSTOM_CONTAINER,
        },
    },
    batch_id="batch-custom-container",
)
get_batch_custom = DataprocGetBatchOperator(
    task_id="get_batch_custom",
    batch_id="batch-custom-container",
)
delete_batch_custom = DataprocDeleteBatchOperator(
    task_id="delete_batch_custom",
    batch_id="batch-custom-container",
)
create_batch_with_custom_container >> get_batch_custom >> delete_batch_custom

Usar o serviço Metastore do Dataproc com o DataprocCreateBatchOperator

Para usar um serviço do metastore do Dataproc em um DAG:

1. Verifique se o serviço de metastore já foi iniciado.

   Para saber como iniciar um serviço do metastore, consulte Ativar e desativar o metastore do Dataproc.

   Para informações detalhadas sobre o operador de lote para criar o do Terraform, consulte PeripheralsConfig.

2. Depois que o serviço do metastore estiver em execução, especifique o nome dele na variável metastore_cluster e a região na variável do Airflow region_name.

3. Use o serviço metastore no DataprocCreateBatchOperator:

create_batch_with_metastore = DataprocCreateBatchOperator(
    task_id="dataproc_metastore",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "metastore_service": METASTORE_SERVICE_LOCATION,
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
    },
    batch_id="dataproc-metastore",
)
get_batch_metastore = DataprocGetBatchOperator(
    task_id="get_batch_metatstore",
    batch_id="dataproc-metastore",
)
delete_batch_metastore = DataprocDeleteBatchOperator(
    task_id="delete_batch_metastore",
    batch_id="dataproc-metastore",
)

create_batch_with_metastore >> get_batch_metastore >> delete_batch_metastore

DataprocDeleteBatchOperator

É possível usar o DataprocDeleteBatchOperator para excluir um lote com base no ID da carga de trabalho.

delete_batch = DataprocDeleteBatchOperator(
    task_id="delete_batch",
    batch_id="batch-create-phs",
)

DataprocListBatchesOperator

DataprocDeleteBatchOperator lista lotes que existem em um determinado project_id e região.

list_batches = DataprocListBatchesOperator(
    task_id="list-all-batches",
)

DataprocGetBatchOperator

O DataprocGetBatchOperator busca uma carga de trabalho em lote específica.

get_batch = DataprocGetBatchOperator(
    task_id="get_batch",
    batch_id="batch-create-phs",
)

A seguir

• Visão geral do Dataproc sem servidor
• Acionar DAGs com funções do Cloud Run