嗨，这个场景我太熟悉了——用Spark写S3上的Hive分区表时，小文件和临时文件满天飞确实头疼。结合你的情况（按month、year、city分区，20亿行数据），给你几个实用的解决方案，帮你实现每个分区对应单个文件，同时清理那些烦人的临时文件：

一、写入前按分区键重分区 + 优化Spark配置 #

这是最直接的事前处理方案，让每个分区对应一个Spark任务，从根源上减少小文件：

1. 对DataFrame按分区字段重分区 #

先根据你的分区键（month, year, city）重新分区，确保每个分区组合对应一个Spark任务：

// 假设你的DataFrame是df，先按分区键重分区
val repartitionedDf = df.repartition($"month", $"year", $"city")

这样每个分区组合的数据会被分配到单独的task中，写入时每个task生成一个文件。

2. 配置Spark优化参数（针对S3和分区写入） #

在写入前设置以下配置，优化S3写入逻辑并避免临时文件残留：

// 优化S3文件提交算法，减少临时文件（关键！）
spark.conf.set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")
// 动态覆盖分区（如果是覆盖写入场景）
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
// 调整shuffle分区数（可设为和实际分区数量一致）
spark.conf.set("spark.sql.shuffle.partitions", 你的分区总数)

3. 写入Hive表 #

用saveAsTable或insertInto写入即可：

// 方法1：用saveAsTable关联已创建的Hive表
repartitionedDf.write
  .mode("overwrite") // 或"append"，根据你的需求选择
  .saveAsTable("your_hive_table_name")

// 方法2：用insertInto（需确保DataFrame字段和表结构完全匹配）
repartitionedDf.write
  .mode("overwrite")
  .insertInto("your_hive_table_name")

二、借助Hive的小文件合并参数（Spark SQL方式） #

如果习惯用Spark SQL写入，可以通过配置Hive的合并参数，让Hive自动合并小文件：

1. 注册临时视图 #

先把你的DataFrame注册成临时视图：

df.createOrReplaceTempView("temp_big_data")

2. 设置Hive合并相关配置 #

// 开启动态分区
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
// 开启map阶段和reduce阶段的文件合并
spark.conf.set("hive.merge.mapfiles", "true")
spark.conf.set("hive.merge.mapredfiles", "true")
// 设置合并后的单个文件大小（比如256MB，可按需调整）
spark.conf.set("hive.merge.size.per.task", "256000000")
// 当分区内平均文件大小小于这个值时触发合并（比如16MB）
spark.conf.set("hive.merge.smallfiles.avgsize", "16000000")

3. 执行INSERT OVERWRITE语句 #

INSERT OVERWRITE TABLE your_hive_table_name PARTITION(month, year, city)
SELECT col1, col2, ..., month, year, city FROM temp_big_data;

Hive会自动帮你把每个分区内的小文件合并成符合大小要求的文件（参数设置合理的话，就能实现每个分区一个文件）。

三、事后合并已有小文件（针对已写入的数据） #

如果已经出现了大量小文件，可以用Hive的CONCATENATE命令事后合并：

-- 合并单个分区的文件
ALTER TABLE your_hive_table_name PARTITION(month='2024', year='05', city='beijing') CONCATENATE;

-- 如果要合并所有分区，可以写脚本遍历所有分区执行上述命令

这个命令适用于ORC、Parquet等列式存储格式的表，会高效地合并分区内的小文件。

额外注意事项 #

• 避免使用coalesce：coalesce只是减少分区数，但不会按分区键重新分配数据，容易导致多个分区的数据混到同一个文件里，不符合你的需求。
• S3存储特性：S3是对象存储，删除文件有延迟，所以尽量从写入阶段避免生成临时文件，mapreduce.fileoutputcommitter.algorithm.version=2这个参数一定要设置，它会直接写入目标目录，而不是先写临时目录再移动，大幅减少临时文件。
• 内存压力：20亿行数据按分区重分区时，如果分区数量极大（比如city有上万个），要注意Spark的内存配置，避免OOM，可以适当调整spark.driver.memory和spark.executor.memory。

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