自动扩缩工作负载资源

借助 Managed Service for Apache Spark 自动扩缩功能，您可以动态扩缩工作负载资源（例如执行器的数量），以便高效地运行 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 的随机节点选择替换为智能算法，该算法可识别最适合首先缩容的节点。此算法会考虑节点的 shuffle 数据大小和空闲时间等因素。
• 可配置的 Spark 安全停用和 shuffle 迁移行为：借助 Autoscaling 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 Autoscaling 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	如果 true，则在运行中的执行器超过 spark.dynamicAllocation.diagnosis.interval 指定时间段内所需执行器数量上限时，系统会记录诊断信息。 诊断包括执行器摘要，其中包含空闲执行器数量和空闲时间百分位数、活跃任务分布、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 错误而导致的阶段重试（请参阅 ExecutorDeadException 导致的 FetchFailedException）。此属性在 Managed Service for Apache Spark Spark 运行时版本 1.1.12 及更高版本以及 2.0.20 及更高版本中可用。	false
spark.scheduler.excludeShuffleSkewExecutors	如果为 true，则避免在重排倾斜执行器（即具有大量重排数据或大量已完成映射任务的执行器）上调度任务。这可以通过缓解 shuffle 倾斜来提高性能。	false

Spark 动态分配配置文件

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

performance：此配置文件通过应用以下默认设置来优化执行时间，使其达到最短：

• 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 输出位置。