自动扩缩工作负载 resource_allocation_properties

本文档提供有关 Managed Service for Apache Spark 自动扩缩的信息。在提交 Spark 工作负载时，Managed Service for Apache Spark 可以动态扩缩工作负载资源（例如执行程序的数量），以便高效地运行工作负载。Managed Service for Apache Spark 自动扩缩是默认行为，它使用 Spark 动态资源分配来确定是否扩缩工作负载、扩缩方式和扩缩时间。

Managed Service for Apache Spark 自动扩缩 V2

Managed Service for Apache Spark 自动扩缩版本 2 (V2) 在默认版本 1 (V1) 的基础上新增了多项功能并进行了改进，可帮助您管理 Managed Service for Apache Spark 工作负载、提升工作负载性能并节省费用：

• 异步节点缩容：自动扩缩 V2 将 V1 的同步缩容替换为异步缩容。使用异步缩容时，Managed Service for Apache Spark 会缩减工作负载资源，而无需等待所有节点完成 Shuffle 迁移。这意味着，缩容速度缓慢的长尾节点不会阻止扩容。
• 智能缩容节点选择：自动伸缩 V2 使用智能算法取代了 V1 的随机节点选择，该算法可识别最适合率先缩容的节点。此算法会考虑节点混洗数据大小和空闲时间等因素。
• 可配置的 Spark 正常退役和 Shuffle 迁移行为：借助自动扩缩 V2，您可以使用标准 Spark 属性来配置 Spark 正常退役和 Shuffle 迁移。此功能可帮助您保持迁移与自定义 Spark 属性的兼容性。

注意： 在提交 Spark 批处理工作负载时，设置 spark.dataproc.scaling.version=2 属性可启用 Managed Service for Apache Spark 自动扩缩 V2。

Managed Service for Apache Spark 自动扩缩功能

功能	Managed Service for Apache Spark 自动扩缩功能 V1	Managed Service for Apache Spark 自动扩缩功能 V2
节点缩减	同步	异步
缩小规模的节点选择	随机	智能
Spark 安全停用和 Shuffle 迁移	不可配置	可配置

Spark 动态分配属性

下表列出了您在提交批处理工作负载时可以设置的 Spark 动态分配属性，以控制自动扩缩（请参阅如何设置 Spark 属性）。

注意： 另请参阅 Spark 属性，其中介绍了在提交 Managed Service for Apache Spark 批处理工作负载时可以设置的其他属性。

属性	说明	默认
spark.dataproc.scaling.version	Managed Service for Apache Spark 自动扩缩版本。指定版本 1 或 2。	1
spark.dynamicAllocation.enabled	是否使用动态资源分配，该功能可根据工作负载扩缩执行程序的数量。 将值设置为 false 会停用工作负载的自动扩缩功能。默认值：true。	true
spark.dynamicAllocation.initialExecutors	分配给工作负载的执行器的初始数量。工作负载开始运行后，自动扩缩可能会更改活跃执行程序的数量。最小值为 2；最大值为 2000。	2
spark.dynamicAllocation.minExecutors	工作负载可缩减到的执行程序数量下限。 最小值为 2。	2
spark.dynamicAllocation.maxExecutors	工作负载可扩容到的执行程序数上限。 值不能超过 2000。	1000
spark.dynamicAllocation.executorAllocationRatio	自定义 Spark 工作负载的纵向伸缩。接受的值介于 0 到 1 之间。值为 1.0 时，可提供最大的纵向扩缩能力，并有助于实现最大并行性。值为 0.5 时，将纵向扩缩能力和并行性设置为最大值的一半。	0.3
spark.dynamicAllocation.diagnosis.enabled	如果运行的执行程序在 spark.dynamicAllocation.diagnosis.interval 指定的时间段内超过了所需的最大执行程序数，系统会记录诊断信息（当值为 true 时）。 诊断信息包括执行程序摘要（包含空闲执行程序数量和空闲时间百分位数）、活跃任务分布、Shuffle 数据大小和缓存的 RDD 大小。使用 spark.dynamicAllocation.diagnosis.logLevel 控制输出日志级别。	false
spark.dynamicAllocation.profile	设置为 performance 或 cost 可应用针对性能或性价比进行优化的预定义配置集。 用户定义的属性会覆盖配置文件的默认值。如需了解详情，请参阅 Spark 动态分配配置文件。	none
spark.dynamicAllocation.shuffleTracking.dynamicTimeout.enabled	当值为 true 时，为持有 shuffle 数据的执行器启用动态超时计算。 超时时间不再使用静态 spark.dynamicAllocation.shuffleTracking.timeout，而是根据存储在执行器上的 shuffle 数据量计算得出。 这样一来，shuffle 数据量较小的执行器可以更快地释放，而 shuffle 数据量较大的执行器可以保持更长时间的活跃状态。	false
spark.reducer.fetchMigratedShuffle.enabled	设置为 true 时，在因 Spark 动态分配而停用的执行器提取失败后，启用从 Spark 驱动程序提取 Shuffle 输出位置。这可减少因 Shuffle 块从已停用的执行器迁移到正在运行的执行器而导致的 ExecutorDeadException 错误，并减少因 FetchFailedException 错误而导致的阶段重试（请参阅 FetchFailedException caused by ExecutorDeadException）。此属性适用于 Managed Service for Apache Spark Spark 运行时版本 1.1.12 及更高版本和 2.0.20 及更高版本。	false
spark.scheduler.excludeShuffleSkewExecutors	当值为 true 时，避免在 shuffle 倾斜的执行器上调度任务，这些执行器具有大量的 shuffle 数据或大量的已完成的 map 任务。这可以通过缓解 shuffle 倾斜来提高性能。	false

Spark 动态分配配置文件

您可以将 spark.dynamicAllocation.profile 属性设置为 performance 或 cost，以应用针对性能或成本效益进行优化的预定义 Spark 配置集。如果您在设置 spark.dynamicAllocation.profile 属性的同时还设置了 Spark 属性，您的设置将替换相应属性的配置文件默认值。

性能：此配置文件通过应用以下默认设置来优化执行时间：

• spark.scheduler.excludeShuffleSkewExecutors：true
• spark.dynamicAllocation.executorIdleTimeout：300s
• spark.dynamicAllocation.initialExecutors：10

cost：此配置文件通过应用以下默认设置来优化资源消耗，以降低费用：

• spark.dynamicAllocation.executorIdleTimeout：120s
• spark.dynamicAllocation.cachedExecutorIdleTimeout：120s
• spark.dynamicAllocation.shuffleTracking.dynamicTimeout.enabled：true
• spark.dynamicAllocation.diagnosis.enabled：true

Spark 动态分配指标

Spark 批量工作负载会生成以下与 Spark 动态资源分配相关的指标（如需详细了解 Spark 指标，请参阅监控和插桩）。

注意： 请参阅 Spark 指标，其中介绍了您可以设置哪些属性来控制 Spark 指标的收集。

指标	说明
maximum-needed	在当前负载下，满足所有正在运行和待处理任务所需的执行程序数上限。
running	正在执行任务的执行程序数量。

Spark 动态分配问题和解决方案

• 由 ExecutorDeadException 引起的 FetchFailedException

  原因：当 Spark 动态分配缩减执行程序时，shuffle 文件会迁移到正在运行的执行程序。不过，由于执行器上的 Spark reducer 任务会在 reducer 任务启动时从 Spark 驱动程序设置的位置提取 shuffle 输出，因此如果迁移了 shuffle 文件，reducer 可能会继续尝试从已停用的执行器提取 shuffle 输出，从而导致 ExecutorDeadException 和 FetchFailedException 错误。

  解决方案：在运行 Managed Service for Apache Spark 批量工作负载时，将 spark.reducer.fetchMigratedShuffle.enabled 设置为 true，以启用 shuffle 位置信息重新提取（请参阅设置 Spark 批量工作负载属性）。 启用此属性后，如果从已停用的执行程序提取数据失败，reducer 任务会从驱动程序重新提取 shuffle 输出位置。