Spark Iceberg插入任务本地排序阶段异常缓慢，性能远低于全局排序

问题概述 #

将数据从Iceberg源表插入到已配置本地排序的Iceberg目标表，执行的表属性修改语句为：

ALTER TABLE schema1.test_iceberg_ordered1 WRITE DISTRIBUTED BY PARTITION LOCALLY ORDERED BY example_event_cd NULLS LAST

实际执行时，本地排序任务的速度比全局排序慢10倍，与“本地排序工作量更低”的预期不符，同时出现磁盘溢出现象，推测Spark配置不合理是核心诱因。

插入SQL语句 #

spark.sql(f""" INSERT INTO Omniture_new.core_page_view_iceberg_ordered1
               SELECT * FROM Omniture_new.core_page_view_iceberg WHERE page_view_dtm BETWEEN TIMESTAMP '2023-05-21 00:00:00' AND TIMESTAMP '2023-05-23 23:59:59' """)

当前Spark配置 #

.config('spark.master','yarn')
.config("spark.executor.instances","20")
.config("spark.executor.cores",30)
.config("spark.executor.memory","8G")
.config("spark.sql.shuffle.partitions",10)
.config("spark.default.parallelism",200)
.config("spark.sql.adaptive.coalescePartitions.parallelismFirst","false")
.config("spark.sql.adaptive.advisoryPartitionSizeInBytes",10737418240)
.config("spark.sql.files.maxPartitionBytes", "24182400")
.config("spark.sql.adaptive.enabled", "true")
.config("spark.driver.memory","16G")

任务执行信息 #

• 任务执行情况截图
• Spark执行计划截图

核心问题诊断与优化方案 #

1. Shuffle分区数严重不足 #

spark.sql.shuffle.partitions=10远低于spark.default.parallelism=200，导致每个shuffle分区需处理超大规模数据，内存无法容纳时频繁触发磁盘溢出，直接拖慢本地排序速度。本地排序依赖单分区内的数据内存排序，分区过大必然导致大量磁盘IO开销。

• 优化建议：将spark.sql.shuffle.partitions调整为与spark.default.parallelism匹配的数值（如200），或根据数据总量估算，保持每个shuffle分区大小在128MB-256MB区间。

2. Executor资源配比失衡 #

每个Executor分配30核但仅8G内存，核内存比约为1:266MB，严重低于Spark推荐的1:3-4G合理比例，内存资源不足是磁盘溢出的直接原因。

• 优化建议：
  • 方案一：降低spark.executor.cores至8-10，同时将spark.executor.memory提升至32G，保证每核内存充足；
  • 方案二：保持20个Executor，将spark.executor.memory上调至24G-32G，直接扩容排序可用内存。

3. 自适应分区配置冲突 #

• spark.sql.adaptive.advisoryPartitionSizeInBytes=10G设置过大，自适应机制会强制合并出超大分区，进一步加剧内存压力；
• spark.sql.files.maxPartitionBytes=24MB过小，读取阶段生成大量小分区，增加任务调度与合并开销。
• 优化建议：
  • 将spark.sql.adaptive.advisoryPartitionSizeInBytes调整为128MB-256MB；
  • 将spark.sql.files.maxPartitionBytes恢复为默认128MB，减少小分区数量。

4. Iceberg本地排序逻辑验证 #

• 确认目标表的分区键与DISTRIBUTED BY PARTITION完全匹配，保证shuffle后数据按分区均匀分布，避免单分区数据量过载；
• 检查example_event_cd字段的数据分布，若存在严重数据倾斜（如大量重复值），需增加分区维度或对排序键做哈希散列优化，均衡各分区排序压力。

5. 内存细节优化 #

• 添加spark.executor.memoryOverhead配置，设置为Executor内存的20%-30%，避免堆外内存不足引发的溢出；
• 确保spark.sql.sort.partitions与spark.sql.shuffle.partitions保持一致，保障排序并行度匹配。

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