Spark本地集群读并行但写分区仅单核心，如何并行提速？

我使用本地Spark集群将300GB CSV文件写入现有Iceberg表，Spark上下文初始化配置如下：

.master("local[*]") \
.config("spark.driver.memory", "30g") \
.config("spark.executor.memory", "15g") \
.config("spark.dynamicAllocation.enabled", "true") \
.config("spark.sql.catalog.my_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
.config("spark.hadoop.fs.s3a.fast.upload", "true") \

经过转换后，我创建了partition_key将DataFrame划分为大小相近的分区，最终S3中的文件大小约为200MB。执行第一阶段（读取、应用schema、转换及分区）时Spark运行高效，20分钟完成2470个任务，处理300GB输入、50GB shuffle写入，且使用全部10个本地核心。

但进入第二阶段（实际写入S3）时，Spark仅使用一个核心进行写入（注：不受网络带宽限制），写入代码如下：

df.repartition("partition_key") \
    .write \
    .format("iceberg") \
    .mode("append") \
    .option("path", "s3://my_bucket/my_db/") \
    .partitionBy(["partition_key"]) \
    .saveAsTable("glue_catalog.my_db.data")

我了解到本地Spark实例本质上只有一个执行器（本地JVM实例），可能限制写入，但读取阶段却能并行使用全部核心，这不符合逻辑。如何加速并并行化分区写入？我已尝试启用动态分配但无效果：

.config("spark.dynamicAllocation.enabled", "true") \

为何Spark默认实现读取并行化，而分区写入却无法并行？

一、写入单核心运行的核心原因 #

1. 本地模式的执行器特性：本地模式(local[*])确实只有一个JVM执行器，但该执行器可利用多个核心（你这里是10个）。读取阶段能并行是因为CSV数据源会自动拆分出大量任务（对应文件块/分区），这些任务可在多核心同时执行；而写入阶段任务数不足或被合并，导致核心闲置。
2. 重复分区操作的冲突：同时使用repartition("partition_key")和partitionBy(["partition_key"])，会让每个partition_key对应一个Spark分区。若某个分区数据量过大，Iceberg写入时会将该分区的所有数据交给单个任务处理，仅占用一个核心。
3. 动态分配无效的本质：动态分配是针对集群模式的扩展配置，本地模式下Spark不会创建额外执行器，因此该配置完全不起作用。

二、并行化写入的优化方案 #

1. 调整分区策略，拆分大任务 #

不要仅按partition_key分区，而是结合数据量指定总分区数，确保每个Spark分区数据量均衡：

# 按partition_key分区的同时，指定总分区数（根据数据量计算，比如对应10个核心的并行度）
df.repartition(100, "partition_key") \
    .write \
    .format("iceberg") \
    .mode("append") \
    .option("path", "s3://my_bucket/my_db/") \
    .partitionBy(["partition_key"]) \
    .saveAsTable("glue_catalog.my_db.data")

这里的100可根据总数据量和单任务期望数据量（比如200MB）计算，确保多核心能同时处理不同Spark分区的写入任务。

2. 配置Iceberg写入并行参数 #

在Spark上下文配置中添加Iceberg专属参数，强制启用并行写入：

# 新增Iceberg并行写入配置
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
.config("spark.sql.iceberg.write.parallelism", "10") \

spark.sql.iceberg.write.parallelism指定写入阶段的并行度，设置为你的本地核心数（10），让Iceberg拆分写入任务到多核心执行。

3. 移除重复分区操作 #

若目标Iceberg表已按partition_key分区，写入时无需重复指定partitionBy，Iceberg会自动按已有规则处理，避免不必要的shuffle和任务合并：

df.repartition(100, "partition_key") \
    .write \
    .format("iceberg") \
    .mode("append") \
    .saveAsTable("glue_catalog.my_db.data")

三、补充说明 #

本地模式下，单个执行器的多线程（核心）可同时处理多个任务。写入阶段单核心运行的本质是任务数不足，而非核心无法被利用。通过调整分区数和Iceberg写入参数，就能让多核心同时参与写入，大幅提升速度。

内容的提问来源于stack exchange，提问作者iskandarblue