Ajuste automático de cargas de trabalho em lote do Spark

Este documento fornece informações sobre a otimização automática de cargas de trabalho em lote do Serverless para Apache Spark. A otimização de uma carga de trabalho do Spark para desempenho e resiliência pode ser desafiadora devido ao número de opções de configuração do Spark e à dificuldade de avaliar o impacto dessas opções numa carga de trabalho. A otimização automática sem servidor para o Apache Spark oferece uma alternativa à configuração manual da carga de trabalho, aplicando automaticamente as definições de configuração do Spark a uma carga de trabalho recorrente do Spark com base nas práticas recomendadas de otimização do Spark e numa análise das execuções da carga de trabalho (denominadas "coortes").

Inscreva-se na otimização automática do Serverless para Apache Spark

Para se inscrever no acesso à versão de pré-visualização do ajuste automático do Serverless para Apache Spark descrito nesta página, preencha e envie o formulário de inscrição de pedido de acesso à pré-visualização do Dataproc. Depois de o formulário ser aprovado, os projetos indicados no formulário têm acesso às funcionalidades de pré-visualização.

Vantagens

A otimização automática do Serverless para Apache Spark pode oferecer as seguintes vantagens:

  • Otimização automática: ajuste automaticamente as configurações do Apache Spark e do processamento em lote do Serverless ineficientes, o que pode acelerar os tempos de execução das tarefas.
  • Aprendizagem histórica: aprenda com execuções recorrentes para aplicar recomendações adaptadas à sua carga de trabalho.

Coortes de ajuste automático

A otimização automática é aplicada a execuções recorrentes (coortes) de uma carga de trabalho em lote.

O nome da coorte que especifica quando envia uma carga de trabalho em lote identifica-a como uma das execuções sucessivas da carga de trabalho recorrente.

A otimização automática é aplicada a grupos com caraterísticas em comum de cargas de trabalho em lote da seguinte forma:

  • A otimização automática é calculada e aplicada à segunda e às coortes subsequentes de uma carga de trabalho. A otimização automática não é aplicada à primeira execução de uma carga de trabalho recorrente, porque a otimização automática sem servidor para o Apache Spark usa o histórico da carga de trabalho para otimização.

  • A otimização automática não é aplicada retroativamente a cargas de trabalho em execução, mas apenas a cargas de trabalho enviadas recentemente.

  • A otimização automática aprende e melhora ao longo do tempo através da análise das estatísticas da coorte. Para permitir que o sistema recolha dados suficientes, recomendamos que mantenha a otimização automática ativada durante, pelo menos, cinco execuções.

Nomes das coortes: uma prática recomendada é usar nomes das coortes que ajudem a identificar o tipo de carga de trabalho recorrente. Por exemplo, pode usar daily_sales_aggregation como o nome da coorte para uma carga de trabalho agendada que executa uma tarefa de agregação de vendas diária.

Cenários de ajuste automático

Quando aplicável, o ajuste automático seleciona e executa automaticamente os seguintes objetivos scenarios para otimizar uma carga de trabalho em lote:

  • Dimensionamento: definições de configuração do dimensionamento automático do Spark.
  • Otimização de junções: definições de configuração do Spark para otimizar o desempenho das junções de transmissão SQL.

Use o Serverless para a otimização automática do Apache Spark

Pode ativar a otimização automática do Serverless para Apache Spark numa carga de trabalho em lote através da Google Cloud consola, da CLI do Google Cloud ou da API Dataproc, ou das bibliotecas de cliente do Google Cloud.

Consola

Para ativar a otimização automática sem servidor para o Apache Spark em cada envio de uma carga de trabalho em lote recorrente, siga estes passos:

  1. Na Google Cloud consola, aceda à página Batches do Dataproc.

    Aceda aos lotes do Dataproc

  2. Para criar uma carga de trabalho em lote, clique em Criar.

  3. Na secção Ajuste automático:

    • Ative o botão Ativar para ativar o ajuste automático para a carga de trabalho do Spark.

    • Coorte: preencha o nome da coorte, que identifica o lote como um de uma série de cargas de trabalho recorrentes. A otimização automática é aplicada à segunda e às cargas de trabalho subsequentes que são enviadas com este nome de coorte. Por exemplo, especifique daily_sales_aggregation como o nome da coorte para uma carga de trabalho em lote agendada que executa uma tarefa de agregação de vendas diária.

  4. Preencha outras secções da página Criar lote, conforme necessário, e, de seguida, clique em Enviar. Para mais informações sobre estes campos, consulte o artigo Envie uma carga de trabalho em lote.

gcloud

Para ativar a otimização automática sem servidor para o Apache Spark em cada envio de uma carga de trabalho em lote recorrente, execute o seguinte comando da CLI gcloud gcloud dataproc batches submit localmente numa janela de terminal ou no Cloud Shell.

gcloud dataproc batches submit COMMAND \
    --region=REGION \
    --cohort=COHORT \
    --autotuning-scenarios=auto  \
    other arguments ...

Substitua o seguinte:

  • COMMAND: o tipo de carga de trabalho do Spark, como Spark, PySpark, Spark-Sql ou Spark-R.
  • REGION: a região onde a sua carga de trabalho em lote vai ser executada.
  • COHORT: o nome da coorte, que identifica o lote como um de uma série de cargas de trabalho recorrentes. A otimização automática é aplicada à segunda e às cargas de trabalho subsequentes que são enviadas com este nome de coorte. Por exemplo, especifique daily_sales_aggregation como o nome da coorte para uma carga de trabalho em lote agendada que executa uma tarefa de agregação de vendas diária.

  • --autotuning-scenarios=auto: ative o ajuste automático.

API

Para ativar a otimização automática sem servidor para o Apache Spark em cada envio de uma carga de trabalho em lote recorrente, envie um pedido batches.create que inclua os seguintes campos:

  • RuntimeConfig.cohort: o nome da coorte, que identifica o lote como um de uma série de cargas de trabalho recorrentes. A otimização automática é aplicada à segunda e às cargas de trabalho subsequentes enviadas com este nome de coorte. Por exemplo, especifique daily_sales_aggregation como o nome da coorte para uma carga de trabalho em lote agendada que executa uma tarefa de agregação de vendas diária.

  • AutotuningConfig.scenarios: especifique AUTO para ativar o ajuste automático na carga de trabalho em lote do Spark.

Exemplo:

...
runtimeConfig:
  cohort: COHORT_NAME
  autotuningConfig:
    scenarios:
    - AUTO
...

Java

Antes de experimentar este exemplo, siga as Javainstruções de configuração no início rápido do Serverless para Apache Spark com bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Serverless para Apache SparkJava.

Para se autenticar no Serverless para Apache Spark, configure as Credenciais padrão da aplicação. Para mais informações, consulte o artigo Configure a autenticação para um ambiente de desenvolvimento local.

Para ativar a otimização automática sem servidor para o Apache Spark em cada envio de uma carga de trabalho em lote recorrente, chame BatchControllerClient.createBatch com um CreateBatchRequest que inclua os seguintes campos:

  • Batch.RuntimeConfig.cohort: o nome da coorte, que identifica o lote como um de uma série de cargas de trabalho recorrentes. A otimização automática é aplicada à segunda e às cargas de trabalho subsequentes enviadas com este nome de coorte. Por exemplo, pode especificar daily_sales_aggregation como o nome da coorte para uma carga de trabalho em lote agendada que executa uma tarefa de agregação de vendas diária.

  • Batch.RuntimeConfig.AutotuningConfig.scenarios: especifique AUTO para ativar o ajuste automático na carga de trabalho em lote do Spark.

Exemplo:

...
Batch batch =
  Batch.newBuilder()
    .setRuntimeConfig(
      RuntimeConfig.newBuilder()
        .setCohort("daily_sales_aggregation")
        .setAutotuningConfig(
          AutotuningConfig.newBuilder()
            .addScenarios(Scenario.AUTO))
    ...
  .build();

batchControllerClient.createBatch(
    CreateBatchRequest.newBuilder()
        .setParent(parent)
        .setBatchId(batchId)
        .setBatch(batch)
        .build());
...

Para usar a API, tem de usar a versão google-cloud-dataproc da biblioteca cliente 4.43.0 ou posterior. Pode usar uma das seguintes configurações para adicionar a biblioteca ao seu projeto.

Maven

<dependencies>
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-dataproc</artifactId>
   <version>4.43.0</version>
 </dependency>
</dependencies>

Gradle

implementation 'com.google.cloud:google-cloud-dataproc:4.43.0'

SBT

libraryDependencies += "com.google.cloud" % "google-cloud-dataproc" % "4.43.0"

Python

Antes de experimentar este exemplo, siga as Pythoninstruções de configuração no início rápido do Serverless para Apache Spark com bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Serverless para Apache SparkPython.

Para se autenticar no Serverless para Apache Spark, configure as Credenciais padrão da aplicação. Para mais informações, consulte o artigo Configure a autenticação para um ambiente de desenvolvimento local.

Para ativar a otimização automática sem servidor para o Apache Spark em cada envio de uma carga de trabalho em lote recorrente, chame BatchControllerClient.create_batch com um lote que inclua os seguintes campos:

  • batch.runtime_config.cohort: o nome da coorte, que identifica o lote como um de uma série de cargas de trabalho recorrentes. A otimização automática é aplicada à segunda e às cargas de trabalho subsequentes enviadas com este nome de coorte. Por exemplo, pode especificar daily_sales_aggregation como o nome do grupo para uma carga de trabalho em lote agendada que executa uma tarefa de agregação de vendas diária.

  • batch.runtime_config.autotuning_config.scenarios: especifique AUTO para ativar o ajuste automático na carga de trabalho em lote do Spark.

Exemplo:

# Create a client
client = dataproc_v1.BatchControllerClient()

# Initialize request argument(s)
batch = dataproc_v1.Batch()
batch.pyspark_batch.main_python_file_uri = "gs://bucket/run_tpcds.py"
batch.runtime_config.cohort = "daily_sales_aggregation"
batch.runtime_config.autotuning_config.scenarios = [
    Scenario.AUTO
]

request = dataproc_v1.CreateBatchRequest(
    parent="parent_value",
    batch=batch,
)

# Make the request
operation = client.create_batch(request=request)

Para usar a API, tem de usar a versão google-cloud-dataproc da biblioteca cliente 5.10.1 ou posterior. Para o adicionar ao seu projeto, pode usar o seguinte requisito:

google-cloud-dataproc>=5.10.1

Airflow

Em vez de enviar manualmente cada coorte de lotes com ajuste automático, pode usar o Airflow para agendar o envio de cada carga de trabalho de lotes recorrente. Para o fazer, chame BatchControllerClient.create_batch com um Batch que inclua os seguintes campos:

  • batch.runtime_config.cohort: o nome da coorte, que identifica o lote como um de uma série de cargas de trabalho recorrentes. A otimização automática é aplicada à segunda e às cargas de trabalho subsequentes enviadas com este nome de coorte. Por exemplo, pode especificar daily_sales_aggregation como o nome da coorte para uma carga de trabalho em lote agendada que executa uma tarefa de agregação de vendas diária.

  • batch.runtime_config.autotuning_config.scenarios: especifique AUTO para ativar o ajuste automático na carga de trabalho em lote do Spark.

Exemplo:

create_batch = DataprocCreateBatchOperator(
    task_id="batch_create",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
        },
        "environment_config": {
            "peripherals_config": {
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
        "runtime_config": {
            "cohort": "daily_sales_aggregation",
            "autotuning_config": {
                "scenarios": [
                    Scenario.AUTO,
                ]
            }
        },
    },
    batch_id="BATCH_ID",
)

Para usar a API, tem de usar a versão google-cloud-dataproc da biblioteca cliente 5.10.1 ou posterior. Pode usar o seguinte requisito do ambiente do Airflow:

google-cloud-dataproc>=5.10.1

Para atualizar o pacote no Cloud Composer, consulte o artigo Instale dependências do Python para o Cloud Composer .

Veja alterações de otimização automática

Para ver as alterações da otimização automática do Serverless para Apache Spark a uma carga de trabalho em lote, execute o comando gcloud dataproc batches describe.

Exemplo: o resultado de gcloud dataproc batches describe é semelhante ao seguinte:

...
runtimeInfo:
  propertiesInfo:
    # Properties set by autotuning.
    autotuningProperties:
      spark.dataproc.sql.broadcastJoin.hints:
        annotation: Converted 1 Sort-Merge Joins to Broadcast Hash Join
        value: v2;Inner,<hint>
      spark.dynamicAllocation.initialExecutors:
        annotation: Adjusted Initial executors based on stages submitted in first
          2 minutes to 9
        overriddenValue: '2'
        value: '9'
      spark.dynamicAllocation.maxExecutors:
        annotation: Tuned Max executors to 11
        overriddenValue: '5'
        value: '11'
      spark.dynamicAllocation.minExecutors:
        annotation: Changed Min executors to 9
        overriddenValue: '2'
        value: '9'
...

Pode ver as alterações de ajuste automático mais recentes que foram aplicadas a uma carga de trabalho em execução, concluída ou com falhas na página Detalhes do lote na Google Cloud consola, no separador Resumo.

Painel de resumo da otimização automática.

Preços

A otimização automática do Serverless para Apache Spark é oferecida durante a pré-visualização privada sem custos adicionais. Aplicam-se os preços do Serverless para Apache Spark padrão.