本文旨在解决 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。
Apache Superset Dashboard and SQL Exploration Skill下载Apache Superset 是一个广泛采用的开源 BI 平台，用于 SQL 探索、图表构建和仪表板交付。当代理需要查询仓库数据、组装仪表板或使用成熟的分析界面解释指标而不是临时笔记本代码时，此技能非常有用。
优化 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 存储桶的性能，以达到最佳的写入性能。
