Spark 并行读取但写入分区时仅使用单核的解决方案

本文针对 Spark 在本地模式下读取 CSV 文件并写入 Iceberg 表时，读取阶段能够并行执行，而写入阶段却只能单核运行的问题，提供了详细的解决方案。通过调整 Spark 配置，例如禁用动态资源分配、显式设置 executor 数量和资源，以及优化 AWS CLI 配置，可以有效提升写入性能，实现并行写入。

在使用 Spark 处理大规模数据时，我们经常会遇到读取速度快，但写入速度慢的问题。尤其是在本地模式下，虽然 Spark 能够利用多核并行读取数据，但在写入分区时，却可能退化为单核运行，导致整体性能瓶颈。本文将针对这一问题，提供详细的解决方案和优化建议。

理解问题根源

在本地模式下，Spark 默认只有一个 Executor，即本地 JVM 实例。虽然读取操作可以利用多线程并行执行，但写入操作往往受限于单个 Executor 的资源限制。动态资源分配（spark.dynamicAllocation.enabled）在这种情况下可能不会起到预期效果，甚至可能将所有任务分配给单个 Executor。

解决方案：显式配置 Executor 资源

避免依赖动态资源分配，而是显式地配置 Executor 的数量和资源，可以有效解决单核写入的问题。

1. 禁用动态资源分配：

   移除 spark.dynamicAllocation.enabled = true 的配置。

2. 显式设置 Executor 数量和资源：

   在提交 Spark 应用时，通过命令行参数设置 Executor 的数量、内存和 CPU 核心数。例如：

   spark-submit \
   --master yarn \
   --deploy-mode cluster \
   --num-executors 4 \
   --executor-memory 1G \
   --executor-cores 1 \
   --driver-memory 2G \
   --driver-cores 1 \
   ...

   • --num-executors: 设置 Executor 的数量。
   • --executor-memory: 设置每个 Executor 的内存大小。
   • --executor-cores: 设置每个 Executor 的 CPU 核心数。

   注意：需要根据集群的资源情况和数据规模，合理调整这些参数。可以通过 Yarn Resource Manager UI 监控 CPU 核心的利用率，进行微调。

3. 监控 Spark History Server UI：

   在写入操作开始时，通过 Spark History Server UI 检查 Executor 的数量和 Task 的数量，确保任务能够均匀分配到各个 Executor 上。

优化 AWS CLI 配置

如果数据写入目标是 AWS S3，还可以通过优化 AWS CLI 的配置，提升写入性能。

1. 增加并行度：

   通过设置 max_concurrent_requests 和 max_queue_size 参数，增加并发请求的数量和任务队列的大小。

   [default]
   s3 =
       max_concurrent_requests = 20
       max_queue_size = 1000

2. 调整分片上传参数：

   通过设置 multipart_threshold 和 multipart_chunksize 参数，优化分片上传的策略。

   [default]
   s3 =
       multipart_threshold = 64MB
       multipart_chunksize = 16MB

3. 限制带宽（可选）：

   如果需要限制上传带宽，可以通过设置 max_bandwidth 参数实现。

   [default]
   s3 =
       max_bandwidth = 100MB/s

   注意：这些参数需要在 AWS CLI 的配置文件中进行设置，通常位于 ~/.aws/config。

总结

通过显式配置 Executor 资源和优化 AWS CLI 配置，可以有效解决 Spark 在本地模式下写入分区时仅使用单核的问题，显著提升写入性能。在实际应用中，需要根据具体的硬件环境和数据规模，进行参数调优，以达到最佳效果。此外，监控 Spark History Server UI 和 Yarn Resource Manager UI，可以帮助我们更好地了解任务的执行情况，并及时发现潜在的问题。