自动调优 Spark 工作负载

本文档介绍了如何自动调整 Spark 工作负载。由于 Spark 配置选项众多，并且难以评估这些选项对工作负载的影响，因此优化 Spark 工作负载以提高性能和弹性可能是一项艰巨的任务。 Google Cloud 无服务器 Apache Spark 自动调优功能可根据 Spark 优化最佳实践和对工作负载运行情况的分析，自动将 Spark 配置设置应用于周期性 Spark 工作负载，从而替代手动工作负载配置。

注册 Google Cloud Serverless for Apache Spark 自动调优

如需注册以获取本页所述的 Serverless for Apache Spark 自动调优预览版的访问权限，请填写并提交 Dataproc 预览版访问权限申请注册表单。表单获得批准后，表单中列出的项目便可使用预览版功能。

优势

Google Cloud Serverless for Apache Spark 自动调优可提供以下优势：

  • 提升性能：通过优化调整来提升性能
  • 更快地进行优化：自动配置，避免耗时的人工配置测试
  • 提高恢复能力：自动分配内存，以避免出现与内存相关的故障

限制

Google Cloud Serverless for Apache Spark 自动调优具有以下限制：

  • 自动调优功能会在工作负载的第二次及后续运行中进行计算并应用。由于 Google Cloud Serverless for Apache Spark 自动调优功能会使用工作负载历史记录进行优化，因此首次运行的周期性工作负载不会自动调优。
  • 不支持内存缩减。
  • 自动调优不会追溯性地应用于正在运行的工作负载，只会应用于新提交的工作负载群组。

自动调节同类群组

自动调优功能适用于批量工作负载的重复执行，称为同类群组。您在提交工作负载时指定的同类群组名称用于将该工作负载标识为周期性工作负载的连续运行之一。我们建议您使用描述工作负载类型的同类群组名称，或以其他方式帮助识别工作负载运行是否属于周期性工作负载。例如，将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已调度工作负载的同类群组名称。

自动调节场景

您可以通过选择以下一个或多个自动调优方案，将 Google Cloud Serverless for Apache Spark 自动调优应用于工作负载：

  • MEMORY：自动调整 Spark 内存分配，以预测和避免潜在的工作负载内存不足错误。修复了之前因内存不足 (OOM) 错误而失败的工作负载。
  • SCALING：自动调整 Spark 自动扩缩配置设置。
  • BROADCAST_HASH_JOIN：自动调整 Spark 配置设置，以优化 SQL 广播联接性能。

价格

Google Cloud 在预览版期间，Serverless for Apache Spark 自动调优可免费提供。您需要按标准 Google Cloud Serverless for Apache Spark 价格付费。

区域级可用性

您可以在支持 Compute Engine 的区域中使用 Serverless for Apache Spark 自动调优功能来处理提交的批处理作业。 Google Cloud

使用 Google Cloud Serverless for Apache Spark 自动调优

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Dataproc API 为工作负载启用 Google Cloud Serverless for Apache Spark 自动调优功能。

控制台

如需在每次提交周期性批量工作负载时启用 Google Cloud 无服务器 Apache Spark 自动调优功能，请执行以下步骤：

  1. 在 Google Cloud 控制台中，前往 Dataproc 批次页面。

    转到 Dataproc 批次

  2. 如需创建批处理工作负载，请点击创建。

  3. 在容器部分，为 Spark 工作负载填写以下字段：

    • 同类群组：用于标识批次是否属于一系列周期性工作负载的同类群组名称。 自动调优功能会应用于使用此同类群组名称提交的第二个及后续工作负载。例如，将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已调度工作负载的同类群组名称。

    • 自动调节场景：用于优化工作负载的一个或多个自动调节场景，例如 BROADCAST_HASH_JOIN、MEMORY 和 SCALING。 您可以在每次提交批次队列时更改方案选择。

  4. 根据需要填写创建批次页面的其他部分，然后点击提交。如需详细了解这些字段，请参阅提交批处理工作负载。

gcloud

如需在每次提交周期性批处理工作负载时启用 Google Cloud 无服务器 Apache Spark 自动调优，请在本地终端窗口或 Cloud Shell 中运行以下 gcloud CLI gcloud dataproc batches submit 命令。

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

替换以下内容：

  • COMMAND：Spark 工作负载类型，例如 Spark、PySpark、Spark-Sql 或 Spark-R。
  • REGION：工作负载将运行的区域。

  • COHORT：同类群组名称，用于将批次标识为一系列周期性工作负载之一。 自动调优功能会应用于使用此同类群组名称提交的第二个及后续工作负载。例如，将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已安排工作负载的同类群组名称。

  • SCENARIOS：一个或多个以英文逗号分隔的自动调优方案，用于优化工作负载，例如 --autotuning-scenarios=MEMORY,SCALING。您可以在每次提交批次队列时更改方案列表。

API

如需在每次提交周期性批量工作负载时启用 Google Cloud Serverless for Apache Spark 自动调优，请提交包含以下字段的 batches.create 请求：

  • RuntimeConfig.cohort：同类群组名称，用于将批次标识为一系列周期性工作负载之一。自动调优功能会应用于使用此同类群组名称提交的第二个及后续工作负载。例如，将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已调度工作负载的同类群组名称。
  • AutotuningConfig.scenarios：用于优化工作负载的一个或多个自动调优方案，例如 BROADCAST_HASH_JOIN、MEMORY 和 SCALING。 您可以在每次提交批次队列时更改方案列表。

示例：

...
runtimeConfig:
  cohort: daily_sales_aggregation
  autotuningConfig:
    scenarios:
    - BROADCAST_HASH_JOIN
    - MEMORY
    - SCALING
...

Java

在尝试此示例之前，请按照《Serverless for Apache Spark 快速入门：使用客户端库》中的 Java 设置说明进行操作。 如需了解详情，请参阅 Serverless for Apache Spark Java API 参考文档。

如需向 Serverless for Apache Spark 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

如需在每次提交周期性批处理工作负载时启用 Google Cloud Serverless for Apache Spark 自动调优，请使用包含以下字段的 CreateBatchRequest 调用 BatchControllerClient.createBatch：

  • Batch.RuntimeConfig.cohort：群组名称，用于将批次标识为一系列周期性工作负载之一。自动调优功能会应用于使用此同类群组名称提交的第二个及后续工作负载。例如，您可以将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已安排工作负载的同类群组名称。
  • Batch.RuntimeConfig.AutotuningConfig.scenarios：用于优化工作负载的一个或多个自动调优方案，例如 BROADCAST_HASH_JOIN、MEMORY、SCALING。您可以在每次提交批次队列时更改方案列表。 如需查看完整的情景列表，请参阅 AutotuningConfig.Scenario Javadoc。

示例：

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

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

如需使用该 API，您必须使用 google-cloud-dataproc 客户端库版本 4.43.0 或更高版本。您可以使用以下配置之一将该库添加到项目中。

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

在尝试此示例之前，请按照《Serverless for Apache Spark 快速入门：使用客户端库》中的 Python 设置说明进行操作。 如需了解详情，请参阅 Serverless for Apache Spark Python API 参考文档。

如需向 Serverless for Apache Spark 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

如需在每次提交周期性批量工作负载时启用 Google Cloud Serverless for Apache Spark 自动调优功能，请调用 BatchControllerClient.create_batch，并提供包含以下字段的 Batch：

  • batch.runtime_config.cohort：群组名称，用于将批次标识为一系列周期性工作负载之一。自动调优功能会应用于使用此同类群组名称提交的第二个及后续工作负载。例如，您可以将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已安排工作负载的同类群组名称。
  • batch.runtime_config.autotuning_config.scenarios：用于优化工作负载的一个或多个自动调优方案，例如 BROADCAST_HASH_JOIN、MEMORY、SCALING。您可以在每次提交批次群组时更改方案列表。如需查看完整的情景列表，请参阅情景参考文档。

示例：

# 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.SCALING
]

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

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

如需使用该 API，您必须使用 google-cloud-dataproc 客户端库版本 5.10.1 或更高版本。如需将其添加到项目中，您可以使用以下要求：

google-cloud-dataproc>=5.10.1

Airflow

如需在每次提交周期性批量工作负载时启用 Google Cloud Serverless for Apache Spark 自动调优功能，请调用 BatchControllerClient.create_batch，并提供包含以下字段的 Batch：

  • batch.runtime_config.cohort：群组名称，用于将批次标识为一系列周期性工作负载之一。自动调优功能会应用于使用此同类群组名称提交的第二个及后续工作负载。例如，您可以将 daily_sales_aggregation 指定为运行每日销售额汇总任务的已安排工作负载的同类群组名称。
  • batch.runtime_config.autotuning_config.scenarios：用于优化工作负载的一个或多个自动调优方案，例如 BROADCAST_HASH_JOIN、MEMORY、SCALING。您可以在每次提交批次群组时更改方案列表。如需查看完整的情景列表，请参阅情景参考文档。

示例：

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.SCALING,
                ]
            }
        },
    },
    batch_id="BATCH_ID",
)

如需使用该 API，您必须使用 google-cloud-dataproc 客户端库版本 5.10.1 或更高版本。您可以使用以下 Airflow 环境要求：

google-cloud-dataproc>=5.10.1

如需更新 Cloud Composer 中的软件包，请参阅为 Cloud Composer 安装 Python 依赖项 。

查看自动调优更改

如需查看 Google Cloud Serverless for Apache Spark 对批量工作负载的自动调优更改，请运行 gcloud dataproc batches describe 命令。

示例：gcloud dataproc batches describe 输出类似于以下内容：

...
runtimeInfo:
   propertiesInfo:
    # Properties set by autotuning.
    autotuningProperties
      spark.driver.memory:
        annotation: Driver OOM was detected
        value: 11520m
      spark.driver.memoryOverhead:
        annotation: Driver OOM was detected
        value: 4608m
    # Old overwritten properties.
    userProperties
...

您可以在 Google Cloud 控制台的批处理详细信息页面上的调查标签页下，查看已应用于正在运行、已完成或失败的工作负载的最新自动调优更改。

自动调优调查面板。