Spark分区单核优化技巧分享

本文针对Spark在本地模式下写入Iceberg表时，遇到的单核写入瓶颈问题，提供了一套优化方案。**Spark写入优化：**主要聚焦于解决Spark读取CSV文件后，写入阶段无法充分利用多核资源的问题。**Spark单核优化：**文章深入剖析了单核写入的常见原因，包括Spark默认资源分配策略、动态资源分配的误用以及AWS CLI配置限制。**Spark性能提升：**通过调整Spark配置，如显式设置Executor数量和资源，以及优化AWS CLI配置，如增加并发请求数量，有效提升Spark写入性能。**Spark并发写入：**此外，本文还提供了一个Spark写入Iceberg表的示例代码，并强调了资源分配、网络带宽、S3存储桶性能以及数据倾斜等注意事项，旨在帮助读者充分利用计算资源，实现Spark写入的并行化，从而显著缩短作业执行时间。

本文旨在解决 Spark 在本地模式下读取 CSV 文件并写入 Iceberg 表时，读取阶段能够充分利用多核并行处理，而写入阶段却只能单核运行的问题。通过调整 Spark 配置、优化 AWS CLI 设置，以及理解 Spark 任务分配机制，帮助读者充分利用计算资源，提升 Spark 写入性能。

在使用 Spark 处理大数据集时，一个常见的瓶颈是写入数据的速度。 尤其是在将数据写入到云存储（如 S3）时，如果写入过程没有充分利用可用的计算资源，会导致整体作业的执行时间显著增加。本文将深入探讨如何优化 Spark 写入操作，使其能够并行运行，从而提高写入速度。

理解问题：单核写入的原因

在本地模式下，即使指定了 local[*] 作为 master，Spark 仍然可能只使用一个核心进行写入操作。这通常是由于以下几个原因：

1. 默认资源分配： Spark 的默认资源分配策略可能将所有任务分配给单个 executor，尤其是在本地模式下。
2. 动态资源分配的误用： 开启 spark.dynamicAllocation.enabled 并不一定能解决问题，因为其资源分配依赖于默认的资源计算器，可能无法充分利用所有可用核心。
3. S3 写入限制： 默认的 AWS CLI 配置可能限制了并发请求的数量，从而导致写入速度受限。

解决方案：优化 Spark 配置

要解决单核写入的问题，需要调整 Spark 的配置，以确保任务能够并行执行。以下是一些建议的配置：

1. 显式设置 Executor 数量和资源： 不要依赖动态资源分配，而是显式设置 executor 的数量、内存和核心数。

   --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 的核心数。

   根据你的集群资源和数据规模，调整这些参数。

2. 检查 Spark UI： 在写入操作开始时，通过 Spark History Server UI 检查 executor 和任务的数量，确保任务已正确分配到多个 executor。

优化 AWS CLI 配置

除了 Spark 配置，AWS CLI 的配置也会影响写入性能。可以通过增加并发请求的数量来提高写入速度。

1. 修改 AWS CLI 配置文件： 修改 ~/.aws/config 文件，增加以下配置：

   [default]
   s3 =
       max_concurrent_requests = 20
       max_queue_size = 1000
       multipart_threshold = 64M
       multipart_chunksize = 16M
       max_bandwidth = 100MB/s

   • max_concurrent_requests: 允许的最大并发请求数。
   • max_queue_size: 任务队列的最大长度。
   • multipart_threshold: 启用分段上传的最小文件大小。
   • multipart_chunksize: 分段上传时每个分段的大小。
   • max_bandwidth: 限制上传和下载的最大带宽。

   根据你的网络环境和 S3 存储桶的性能，调整这些参数。

代码示例

以下是一个 Spark 写入 Iceberg 表的示例代码，包含了上述优化建议：

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("IcebergWrite") \
    .master("yarn") \
    .config("spark.executor.memory", "1G") \
    .config("spark.executor.cores", "1") \
    .config("spark.num.executors", "4") \
    .config("spark.sql.catalog.my_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.hadoop.fs.s3a.fast.upload", "true") \
    .getOrCreate()

# 读取 CSV 文件
df = spark.read.csv("s3://my_bucket/my_data.csv", header=True, inferSchema=True)

# 数据转换和分区
df = df.withColumn("partition_key", df["some_column"] % 10)  # 示例分区键

# 写入 Iceberg 表
df.repartition("partition_key") \
    .write \
    .format("iceberg") \
    .mode("append") \
    .partitionBy(["partition_key"]) \
    .saveAsTable("glue_catalog.my_db.data")

# 停止 SparkSession
spark.stop()

注意事项

• 资源分配： 确保集群有足够的资源来满足你配置的 executor 数量和资源需求。
• 网络带宽： 如果网络带宽是瓶颈，即使增加了并发请求的数量，写入速度也可能不会显著提高。
• S3 存储桶性能： S3 存储桶的性能也会影响写入速度。如果存储桶的写入能力有限，那么优化 Spark 和 AWS CLI 配置的效果也会受到限制。
• 数据倾斜： 确保分区键能够均匀地分配数据，避免数据倾斜导致某些 executor 负载过重。

总结

通过显式设置 Spark executor 的数量和资源、优化 AWS CLI 配置，以及理解 Spark 任务分配机制，可以有效地解决 Spark 写入分区时仅使用单核的问题，从而提高写入速度，缩短整体作业的执行时间。 在实际应用中，需要根据具体情况调整配置参数，并监控 Spark UI 和 S3 存储桶的性能，以达到最佳的写入性能。

到这里，我们也就讲完了《Spark分区单核优化技巧分享》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！