Spark将S3多分区小Parquet文件合并为仅按part2分区的大文件

嘿，我来帮你搞定这个Parquet文件合并重分区的需求！下面是完整的解决方案，从数据读取到最终写入的全流程优化：

1. 正确读取多层分区数据 #

你一开始用basePath的思路是对的，这能让Spark自动识别路径里的分区列。我给你完善下读取代码，确保能覆盖所有三层分区的小文件：

# 读取所有part1、part2、part3分区的Parquet数据
# basePath指定根目录后，Spark会自动把partX=value解析为DataFrame的列
df = spark.read \
    .option("basePath", "s3://some_bucket/base/location/in/s3/") \
    .parquet("s3://some_bucket/base/location/in/s3/part1=*/*/*")

2. 合并小文件并按part2重分区写入 #

接下来要做的是剔除part1、part3的分区层级，只保留part2作为新分区，同时合并小文件、启用压缩。这里分两种场景给你适配：

场景1：每个part2分区生成单个大文件（适合数据量适中的情况） #

# 先按part2分区，再强制每个分区输出一个文件（避免shuffle的话可以用coalesce）
df_processed = df.repartition("part2")

# 写入S3，配置压缩和分区规则
df_processed.write \
    .mode("overwrite")  # 根据需求选append/overwrite
    .option("compression", "snappy")  # 推荐snappy，平衡压缩率和读写速度
    .option("mergeSchema", "true")  # 不同分区schema有差异时开启
    .partitionBy("part2") \
    .parquet("s3://some_bucket/new/merged/location/")

场景2：控制全局文件数量（适合超大数据量，避免单个文件过大） #

# 先按part2分区，再合并到指定数量的文件（比如20个）
df_processed = df.repartition(20, "part2")

# 写入配置和上面一致
df_processed.write \
    .mode("overwrite")
    .option("compression", "snappy")
    .partitionBy("part2")
    .parquet("s3://some_bucket/new/merged/location/")

3. 关键优化细节 #

• 压缩算法选择：除了snappy，还可以用gzip（压缩率更高但读写慢）、lz4（读写最快），根据你的存储成本和访问速度需求选。
• 减少shuffle开销：如果原数据已经按part2有一定聚集性，用coalesce代替repartition能避免不必要的数据 shuffle，提升性能。
• S3权限配置：确保Spark集群有S3的读写权限，若用s3a协议，需提前配置fs.s3a.access.key、fs.s3a.secret.key等参数。

4. 结果验证 #

写入完成后，去新的S3路径检查：应该只有part2=xxx的目录层级，每个目录下是合并后的大Parquet文件，part1和part3的分区层级已经被移除。

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