开始使用 GPU 加速的 Apache Spark 3

我们在第 3 章讨论了 Spark 3.x 中的 GPU 加速功能。本章将介绍用于 Apache Spark 3 的新 RAPIDS 加速器的使用入门基础知识，以便您利用 GPU 通过 RAPIDS 库加快处理速度（详情请参阅用于 Apache Spark 的 RAPIDS 加速器使用入门）。

用于 Apache Spark 的 RAPIDS 加速器具有以下功能和限制：

支持在 GPU 上通过列式处理运行 Spark SQL
不需要用户更改 API
支持行与列之间的转换处理
使用 RAPIDS cuDF 库
在 GPU 上运行支持的 SQL 操作，如果未执行某项操作或操作不兼容 GPU，它会回退以使用 Spark CPU 版本。
插件无法加速直接操控 RDD 的操作。
加速器库还提供了 Spark shuffle 操作的具体实现，该操作可以利用 UCX 优化 GPU 数据传输（尽可能多地将数据保留在 GPU 上，并绕过 CPU 执行 GPU 到 GPU 传输）。

要启用此 GPU 加速功能，您将需要：

Apache Spark 3.0+
配置有 GPU 的 Spark 集群，而且这些 GPU 符合 RAPIDS Dataframe 库 cuDF 的版本要求。
- 每个执行程序一个 GPU。
添加以下 jar：
- 与集群上可用的 CUDA 版本相对应的 cudf jar。
- RAPIDS Spark 加速器插件 jar。
将配置 spark.plugins 设置为 com.nvidia.spark.SQLPlugin

安装和配置

安装和设置 Spark 以及用于 Spark 的 RAPIDS 加速器时，所需执行的步骤受 Spark 部署方式的影响。部署 Spark 的主要方式如下：

本地模式：驱动程序和任务在同一 Java 虚拟机中运行。此模式仅适用于开发和测试，不宜运行生产应用程序。
在配备集群管理器的集群中：
- Standalone 部署模式：在私有集群中部署 Spark 的至简方式
- Hadoop YARN
- Kubernetes

安装

安装时，您需要在每个工作节点上提供 Spark 3.x、用于 Spark 的 RAPIDS 加速器 jar 文件和 GPU 发现脚本。对于本地模式，您需要在本地安装这些程序。对于 Spark Standalone 模式，您需要在要使用的所有节点上安装这些程序。对于 Yarn 模式，您需要在启动程序节点上安装这些程序，YARN 会根据需要将这些程序传输到相关节点。对于 Kubernetes 模式，您既可将所需程序放入 Docker 映像，也可将其放到运行 Spark 应用程序时所挂载的驱动器上。有关安装的详细信息，请参阅开始使用用于 Apache Spark 的 RAPIDS 加速器。

配置

Spark shell 和 ./bin/spark-submit 通过 --conf 等命令行选项，或通过从 conf/spark-defaults.conf 读取配置选项来支持动态加载配置属性。（请参阅 Spark 配置指南，了解有关 Spark 配置的概述和详情。）

On startup use: --conf [conf key]=[conf 值]。例如：

${SPARK_HOME}/bin/spark --jars 'rapids-4-spark_2.12-0.1.0.jar,cudf-0.14.jar' \
--conf spark.plugins=com.nvidia.spark.SQLPlugin \
--conf spark.rapids.sql.incompatibleOps.enabled=true

在运行时使用：spark.conf.set("[conf 键]", [conf 值])。例如：

scala> spark.conf.set("spark.rapids.sql.incompatibleOps.enabled", true)

GPU 调度

您可以使用 --conf 键值对请求 GPU 并将其分配给相关任务。您使用的具体配置因集群管理器而异。分配 GPU 时所用的几个配置键值属性如下：

请求您的执行程序拥有 GPU：

--conf spark.executor.resource.gpu.amount=1

指定每个任务的 GPU 数量：

--conf spark.task.resource.gpu.amount=1

指定 discoveryScript（在 YARN 和 K8S 上需要）：

--conf spark.executor.resource.gpu.discoveryScript=./getGpusResources.sh

Note that spark.task.resource.gpu.amount can be a decimal amount, so if you want multiple tasks to be run on an executor at the same time and assigned to the same GPU you can set this to a decimal value less than 1. You would want this setting to correspond to the spark.executor.cores setting. For instance, if you have spark.executor.cores=2 which would allow 2 tasks to run on each executor and you want those 2 tasks to run on the same GPU then you would set spark.task.resource.gpu.amount=0.5.

调优

以下配置是建议使用的入门配置，但必须根据具体集群和应用程序进行配置：

Run with one Executor per GPU. Do not try to run with multiple GPUs per executor. Each executor can run with multiple tasks, this depends on the number of cores and number of GPUs on each of your boxes. Run one executor per GPU and you can evenly divide your cores among your executors. For instance, if you have 24 cores and four GPUs per host, you can run with six cores (--conf spark.executor.cores=6)。这样可以控制 Spark 在执行程序上一次加入的任务数量。若要控制 GPU 上一次同时运行的任务数量，您可以配置 spark.rapids.sql.concurrentGpuTasks。一开始最好允许两个任务同时在 GPU 上运行：

(--conf spark.rapids.sql.concurrentGpuTasks=2)，
如果遇到显存不足或性能缓慢问题，则将其更改为 1。造成这种差异的原因在于这类任务仍可使用 CPU，而其他任务则是在 GPU 上运行。目前在 GPU 上一次运行多个任务不会带来性能优势，而且每个任务都会消耗内存，因此我们通常会限制 GPU 上一次运行的任务数量。
设置输入大小。在 GPU 上运行大批数据时，您会发现性能有所提升。但是，输入大小取决于所读取文件的类型以及所执行的操作。
- 如果使用 Spark 数据源 api (spark.read…)，请使用：
- 如果使用 Spark/Hive api 读取 Hive 表中的数据，请使用：
配置 spark.sql.shuffle.partitions 的数量。Spark 设置的默认值为 200，很多时候此设置会导致分区很小。若想增加每个分区的数据大小，以提高 GPU 的处理效率，请尝试尽可能减少分区的数量。您可根据应用程序数据调整此设置以及输入大小。
如果将 KryoSerializer 用于 Spark

(--conf spark.serializer=org.apache.spark.serializer.KryoSerializer)

则必须注册 GpuKryoRegistrator 类，例如：

--conf spark.kryo.registrator=com.nvidia.spark.rapids.GpuKryoRegistrator

配置执行程序内存量，就像为普通 Spark 应用程序配置内存量一样。

一般建议

与使用大量小文件相比，使用少量的大型输入文件效果更佳。您可能无法控制输入文件的大小，但有必要了解一下。
Larger input sizes spark.sql.files.maxPartitionBytes=512m 通常效果更好，前提是数据大小适合 GPU。
The GPU does better with larger data chunks as long as they fit into memory. When using the default spark.sql.shuffle.partitions=200 时，设置较小的分区或许有益。此设置需基于任务中所读取的数据量。每个任务的数据块大小从 512MB 开始设置。

高级配置

除以上配置以外，我们还提供其他特定于插件的配置；只要符合某些要求，这些配置就有助于提高性能。这些配置可控制能够在 GPU 上运行的操作（参见下表）。启用这些配置有助于优化并在 GPU 上运行更多对象，但您务必要了解这些配置的具体功能。例如，GPU 无法与 CPU 版本完全兼容。又如，浮点数可能略有不同。有关配置的更多详细信息，请参阅用于 Spark 的 RAPIDS 加速器配置。

使用物理规划进行监控

用于 Spark 的 RAPIDS 加速器不需要用户更改 API，而且它会将支持的 SQL 操作替换为 GPU 操作。为了解已将哪些操作替换为 GPU 操作，您可以调用 explain 方法打印出特定 DataFrame 的物理规划，其中所有以 GPU 为前缀的操作均是在 GPU 上执行。

现在，我们针对第 1 章介绍的某些查询来比较通过 GPU 处理的特定 DataFrame 的物理规划。在下方的物理规划中，DAG 包含 GpuBatchScan、针对 hour 的 GpuFilter 以及针对 hour、fare_amount 和 day_of_week 的 GpuProject（选择列）。使用 CPU 处理时，DAG 包含 FileScan、Filter 和 Project。

// select 和 filter 是窄依赖 transformation
df.select($"hour", $"fare_amount").filter($"hour" === "0.0" ).show(2)

result:
+----+-----------+
|hour|fare_amount|
+----+-----------+
| 0.0| 10.5|
| 0.0| 12.5|
+----+-----------+

df.select($"hour", $"fare_amount").filter($"hour" === "0.0" ).explain

result:
== Physical Plan ==
*(1) GpuColumnarToRow false<
+- !GpuProject [hour#10, fare_amount#9]
   +- GpuCoalesceBatches TargetSize(1000000,2147483647)
      +- !GpuFilter (gpuisnotnull(hour#10) AND (hour#10 = 0.0))
      +- GpuBatchScan[fare_amount#9, hour#10] GpuCSVScan Location:
InMemoryFileIndex[s3a://spark-taxi-dataset/raw-small/train], ReadSchema: struct<fare_amount:double,hour:double>

请注意，原始计划中的大多数节点如何被 GPU 版本替换。RAPIDS 加速器插入数据格式转换节点（如 GpuColumnarToRow 和 GpuRowToColumnar），以在两种处理之间转换：将在 GPU 上执行的节点的列式处理和将在 CPU 上执行的节点的行处理。为了解某部分查询未在 GPU 上运行的原因，您可将 spark.rapids.sql.explain 配置设置为 true。系统会将此输出记录到驱动日志中，或以交互模式将其显示在屏幕上。

使用 Spark Web UI 进行监控

“SQL”选项卡

了解 GPU 上所运行对象的至简方式是在 Spark Web UI 中查看“SQL”选项卡。在“SQL”选项卡有关下方查询的 DAG 图中，我们看到物理规划中包含 GPUBatchScan、Project、GPUHashAggregate 和 GPUHashAggregate。使用 CPU 处理时，Spark 先对每个分区执行哈希聚合，然后再混洗 Exchange 中的数据以进行宽依赖 transformation 操作。Exchange 之后是对先前的子聚合执行哈希聚合。请注意，GPU 处理期间无需执行 Exchange shuffle 操作。

val df3 = df2.groupBy("month").count
.orderBy(asc("month"))show(5)

“Stages”（阶段）选项卡

您可以通过阶段详情页面查看阶段详情 DAG，其中蓝顶框（方框）表示 RDD 或 DataFrame，边缘（方框之间的箭头）表示对 DataFrame 执行的操作。

“Environment”（环境）选项卡

您可以使用“Environment”（环境）选项卡查看并确认 GPU 配置的属性设置是否正确，例如 Spark.executor.resource.gpu.amount 和 spark.executor.resource.gpu.discoveryScript 属性。您还可在此查看系统属性类路径条目，确认插件 jar 文件是否位于 JVM 类路径下。

表 1. Spark 属性

名称	值
spark.executor.resource.gpu.amount	1
spark.executor.resource.gpu.discoveryScript	/home/ubuntu/getGpusResources.sh

“Executors”（执行程序）选项卡

您可以使用“Executors”（执行程序）选项卡查看已分配给应用程序执行程序的资源。此实例中已分配一个 GPU。

调试

For now, the best way to debug is how you would normally do it on Spark. Look at the UI and log files to see what failed. If you have a seg fault from the GPU find the hs_err_pid.log file. To make sure your hs_err_pid.log file goes into the YARN application log directory, you can add in the config: --conf spark.executor.extraJavaOptions="-XX:ErrorFile=<LOG_DIR>/hs_err_pid_%p.log"。

If you want to see why an operation did not run on the GPU, turn on the configuration: --conf spark.rapids.sql.explain=NOT_ON_GPU。随后系统会在驱动日志中输出一条日志消息，说明 Spark 操作未能在 GPU 上执行的原因。

GPU 显存不足

系统会通过多种方式反映 GPU 显存不足的问题。或许是显示显存不足错误，也可能表现为崩溃问题。通常这意味着分区过大，因此需要返回配置部分调整分区大小和/或分区数量。您也可减少 GPU 并发任务的数量，将其设置为 1。Spark UI 可能会提供有关数据大小的信息。您可以针对失败的阶段查看输入数据或 shuffle 数据的大小。

总结

本章介绍了 Apache Spark 3.x 全新 RAPIDS API 插件的使用入门基础知识，可助您利用 GPU 加快处理速度。有关更多信息，请参阅用于 Spark 的 RAPIDS 加速器指南。