本文旨在解决 Spark 在本地模式下读取 CSV 文件并写入 Iceberg 表时，读取阶段能够充分利用多核并行处理，而写入阶段却只能单核运行的问题。通过调整 Spark 配置、优化 AWS CLI 设置，以及理解 Spark 任务分配机制，帮助读者充分利用计算资源，提升 Spark 写入性能。
在使用 Spark 处理大数据集时，一个常见的瓶颈是写入数据的速度。 尤其是在将数据写入到云存储（如 S3）时，如果写入过程没有充分利用可用的计算资源，会导致整体作业的执行时间显著增加。本文将深入探讨如何优化 Spark 写入操作，使其能够并行运行，从而提高写入速度。
理解问题：单核写入的原因
在本地模式下，即使指定了 local[*] 作为 master，Spark 仍然可能只使用一个核心进行写入操作。这通常是由于以下几个原因：
- 默认资源分配： Spark 的默认资源分配策略可能将所有任务分配给单个 executor，尤其是在本地模式下。
- 动态资源分配的误用： 开启 spark.dynamicAllocation.enabled 并不一定能解决问题，因为其资源分配依赖于默认的资源计算器，可能无法充分利用所有可用核心。
- S3 写入限制： 默认的 AWS CLI 配置可能限制了并发请求的数量，从而导致写入速度受限。
解决方案：优化 Spark 配置
要解决单核写入的问题，需要调整 Spark 的配置，以确保任务能够并行执行。以下是一些建议的配置：
-
显式设置 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 的核心数。
根据你的集群资源和数据规模，调整这些参数。
检查 Spark UI： 在写入操作开始时，通过 Spark History Server UI 检查 executor 和任务的数量，确保任务已正确分配到多个 executor。
优化 AWS CLI 配置
除了 Spark 配置，AWS CLI 的配置也会影响写入性能。可以通过增加并发请求的数量来提高写入速度。
-
修改 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 存储桶的性能，以达到最佳的写入性能。
