Scala Spark写入S3 Parquet因百万级唯一ID分区导致写入缓慢求助

兄弟，你这情况完全是踩了Spark分区的经典大坑——用高基数字段（130多万唯一ID）做分区键，直接导致任务数爆炸，S3的请求开销拉满，写入速度自然慢到离谱。我来给你拆解问题，再给几个实用的解决方案：

问题根源 #

Spark的partitionBy会给每个唯一分区值创建一个独立的任务和输出目录，130多万个ID就意味着130多万个任务要同时和S3交互，光是元数据的创建、小文件的上传就会把集群和S3的带宽/请求配额耗干，更别说数据本身的处理了。而且这么多小文件后续查询时也会严重拖慢性能——Spark要打开几十万甚至上百万个文件，开销比查询数据本身还大。

最优解决方案：用分桶（Bucketing）替代分区 #

分桶是把数据按哈希算法拆分到固定数量的桶里，既能保证按ID查询时的高效性（Spark可以直接定位到对应桶），又能严格控制输出文件的数量。比如你可以设置1000个桶，这样不管有多少个唯一ID，最终只会生成1000个左右的Parquet文件（每个桶一个，或按文件大小拆分几个）。

示例代码：

// 按ID分桶1000个，同时按ID排序（进一步优化查询性能）
output.write
  .mode("append")
  .bucketBy(1000, "id")
  .sortBy("id")
  .parquet("s3://your-target-path")

分桶的优势 #

• 文件数量可控：不用再担心百万级小文件的问题，推荐设置桶数为集群核心数的2-4倍，或者让每个桶的文件大小在50-200MB之间（适配S3存储特性）。
• 查询高效：当你按ID查询时，Spark会计算ID对应的桶，只读取该桶的文件，避免全表扫描。
• 适配高基数字段：完美解决ID这类唯一值极多的字段的分区痛点。

备选方案：分区+分桶结合 #

如果业务上需要保留其他分区维度（比如按日期dt分区），可以把低基数字段作为分区键，高基数的ID作为分桶键。比如按日分区，每个日期分区下设置1000个桶：

output.write
  .mode("append")
  .partitionBy("dt") // 低基数的日期分区
  .bucketBy(1000, "id") // 高基数的ID分桶
  .sortBy("id")
  .parquet("s3://your-target-path")

这样每日的输出文件数就是1000个，远低于百万级，写入和后续查询的性能都会大幅提升。

额外优化技巧 #

1. 控制单文件大小：设置spark.sql.files.maxRecordsPerFile参数，比如设为1000000（100万条记录/文件），让每个Parquet文件的大小稳定在50-100MB左右，避免过小或过大的文件。
2. 优化S3写入：开启S3快速上传，在Spark配置里添加：

   spark.hadoop.fs.s3a.fast.upload=true
   spark.hadoop.fs.s3a.connection.maximum=100 // 增加S3连接数
   

3. 调整Shuffle分区数：如果写入前有shuffle操作，把spark.sql.shuffle.partitions设置为和分桶数一致（比如1000），减少不必要的shuffle开销。
4. 避免频繁Append：如果是每日增量写入，尽量攒够一定量的数据再写入，或者用INSERT INTO替代Append模式（如果是管理表的话），减少小文件的生成。

后续小文件清理（如果已经生成） #

如果已经有大量小文件在S3上，可以用Spark的repartition合并：

spark.read.parquet("s3://existing-path")
  .repartition(1000, "id")
  .write.mode("overwrite")
  .parquet("s3://cleaned-path")

或者用Hive的ALTER TABLE your_table REPARTITION 1000来合并文件（如果是注册为Hive表的话）。

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