如何使用Spark快速将大量大型Parquet文件写入Iceberg？

一、先修正代码中的明显错误 #

你提供的代码存在重复写入的逻辑错误，saveAsTable(table);之后又调用了.writeTo(table).append();，这会导致不必要的重复操作，甚至引发执行逻辑异常，先修正为正确的写入代码：

修正后的代码示例（Scala）： #

spark.read.parquet(path)
  .repartition(500)
  .writeTo(table)
  .append()

或者使用DataFrame API的写法：

spark.read.parquet(path)
  .repartition(500)
  .write.format("iceberg")
  .mode(SaveMode.Append)
  .option("mergeSchema", "true")
  .saveAsTable(table)

注意：二选一即可，不要同时执行两种写入逻辑

二、解决单任务写入的核心问题 #

始终只有一个任务写入，通常是Iceberg写入配置或Spark与Iceberg集成配置缺失导致的，针对你的版本（Spark3.2.0 + Iceberg1.2.1），按以下步骤排查优化：

1. 检查Iceberg Catalog配置是否正确 #

确保Spark Session初始化时配置了Iceberg的扩展和Catalog，这是Iceberg并行写入的基础，示例配置：

--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
--conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
--conf spark.sql.catalog.spark_catalog.type=hive \
# 若使用Hadoop Catalog，替换为：
# --conf spark.sql.catalog.hadoop_catalog=org.apache.iceberg.hadoop.HadoopCatalog \
# --conf spark.sql.catalog.hadoop_catalog.warehouse=hdfs://path/to/warehouse

未正确配置Catalog时，Iceberg可能会降级为单任务写入模式

2. 调整Iceberg写入的并行度配置 #

添加以下Spark配置，强制Iceberg使用多任务并行写入：

--conf spark.sql.iceberg.write.distribution-mode=hash \
--conf spark.sql.iceberg.write.target-file-size-bytes=134217728 \ # 128MB，可根据集群调整
--conf spark.sql.iceberg.write.max-file-size-bytes=268435456 \ # 256MB

• distribution-mode=hash：基于数据哈希值分配任务，保证写入并行性
• 合理设置目标文件大小：避免生成过小文件，同时保证每个任务处理的数据量适配executor内存

3. 优化Spark分区与资源匹配 #

你的spark.sql.shuffle.partitions=500和repartition(500)配置匹配，但需确保资源能支撑并行任务：

• 当前15个executor、每个4核，总核数60，500个任务会排队执行，可适当增加executor数量（比如提升至25个，总核数100），或调整repartition数量为总核数的倍数（200-300），减少任务调度开销
• 默认spark.task.cpus=1（每个核跑一个任务），无需额外修改

4. 排查数据倾斜问题 #

若原始Parquet文件存在严重数据倾斜（单个key行数占比极高），会导致个别任务数据量过大，看起来像单任务执行：

• 通过Spark UI的Stages页面查看各任务的输入数据量，确认是否有任务处理远大于其他任务的数据
• 若存在倾斜，可在repartition时指定多个字段，或用加盐（salting）方式分散倾斜数据

5. 启用Iceberg写入优化特性 #

针对大文件写入，开启批量写入与高效压缩：

--conf spark.sql.iceberg.write.batch-size=10000 \ # 批量提交的记录数
--conf spark.sql.iceberg.write.parquet.compression-codec=zstd \ # 用Zstd压缩降低IO量

Zstd压缩比高于默认的Snappy，能有效减少写入时的IO带宽占用，提升整体速度

三、其他注意事项 #

• 确保存储系统（HDFS等）有足够IO带宽，支撑500GB数据的并行写入吞吐量
• 若Parquet文件Schema与Iceberg表Schema完全一致，去掉mergeSchema=true选项，减少Schema检查开销
• 条件允许可升级Iceberg版本：Iceberg 1.2.1相对老旧，后续版本（如1.4+）对Spark3.2的写入并行度有更多优化

内容的提问来源于stack exchange，提问作者Xiao-Long Li