Spark DataFrame repartition(1)后生成多文件的原因及解决方法

为什么DataFrame 1会生成2个文件？ #

结合你的代码和场景，可能的原因包括：

• 记录数触发文件拆分阈值：如果你的Spark环境设置了spark.sql.files.maxRecordsPerFile参数（控制单个输出文件的最大记录数），当DataFrame 1的记录数超过该阈值时，Spark会自动拆分文件。而DataFrame 2、3的记录数未达到该阈值，因此只生成一个文件。
• Task重试导致残留文件：写入S3时若遇到短暂网络波动等异常，Spark会重试Task，重试过程中可能生成新文件，旧的临时文件若未被正确清理，就会遗留下来形成多个文件。
• 旧Spark版本的兼容性问题：部分早期Spark版本在repartition(1)与partitionBy结合使用时，存在分区处理的潜在bug，可能导致单分区目录下生成多文件。

如何确保每个分区仅生成1个文件？ #

针对上述问题，可采用以下几种可靠方案：

方案1：按分区字段repartition（最优解） #

全局repartition(1)会将所有数据集中到一个分区，既容易引发内存压力，也无法保证单分区文件唯一性。正确的做法是直接按date_column进行repartition，让每个唯一的date_column值对应一个Spark分区：

def writeOneCsvFile(df: DataFrame, s3Location: String) = {
  df.repartition($"date_column")
    .write
    .partitionBy("date_column")
    .format("csv")
    .option("header", true)
    .option("quoteAll", true)
    .save(s3Location)
}

这样每个分区目录下的文件数必然等于Spark分区数（即1个），完美匹配你的需求。

方案2：禁用文件记录数限制 #

检查Spark配置，若存在spark.sql.files.maxRecordsPerFile的非零设置，将其改为0（表示无限制），避免单分区数据被强制拆分：

// 代码中设置，或提交作业时通过--conf spark.sql.files.maxRecordsPerFile=0指定
spark.conf.set("spark.sql.files.maxRecordsPerFile", 0)

方案3：使用coalesce合并分区（谨慎使用） #

如果你的DataFrame原本分区数较少，可使用coalesce(1)替代repartition(1)——它是窄依赖操作，不会触发Shuffle，但如果数据量极大，单个Task处理全量数据可能引发OOM：

def writeOneCsvFile(df: DataFrame, s3Location: String) = {
  df.coalesce(1)
    .write
    .partitionBy("date_column")
    .format("csv")
    .option("header", true)
    .option("quoteAll", true)
    .save(s3Location)
}

方案4：排查Task重试原因 #

若多文件是Task重试导致，建议检查S3网络连接稳定性，或调整Spark的Task重试参数（如spark.task.maxFailures），减少不必要的重试，同时确保作业结束后清理目录下的临时文件（文件名以.tmp开头）。

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