如何限制单个分区内每个Parquet文件的大小至200MB以内

你的代码中repartition("TimeID").write.partitionBy("TimeID")会导致每个TimeID分区仅生成一个Parquet文件——因为repartition("TimeID")已经将数据按TimeID分成了独立的RDD分区，后续partitionBy只是在文件系统层面按目录划分，每个目录对应一个RDD分区的输出，自然只有一个文件。

要实现每个TimeID目录下的Parquet文件不超过200MB，可通过以下几种方案：

方案1：添加随机子分区键（快速实现） #

通过在repartition时额外引入一个随机分区维度，将同一个TimeID的数据拆分为多个RDD分区，从而在同一目录下生成多个文件。你可以根据预期的文件大小估算子分区数量（比如1GB数据拆为5个分区，每个约200MB）：

val subPartitionNum = 5 // 根据单分区最大数据量调整，比如1GB设为5
df.repartition(col("TimeID"), (rand() * subPartitionNum).cast("int"))
  .write.partitionBy("TimeID")
  .parquet("/path/")

方案2：动态计算分区数（精确控制） #

如果需要更精确的文件大小控制，可以先统计每个TimeID的大致数据量，再动态计算每个TimeID需要拆分的分区数：

// 1. 预统计每个TimeID的近似数据量（用JSON序列化长度估算字节数）
val timeIdSizeMap = df.groupBy("TimeID")
  .agg(sum(length(to_json(struct(df.columns.map(col):_*)))).alias("approx_bytes"))
  .collect()
  .map(row => (row.getAs[String]("TimeID"), row.getAs[Long]("approx_bytes")))
  .toMap

// 2. 定义UDF，根据TimeID的大小返回子分区索引
val maxFileSize = 200 * 1024 * 1024L // 200MB字节数
val getSubPartition = udf((timeId: String) => {
  val totalBytes = timeIdSizeMap.getOrElse(timeId, 0L)
  val partitionCount = Math.max(1, Math.ceil(totalBytes.toDouble / maxFileSize).toInt)
  (math.random() * partitionCount).toInt
})

// 3. 按TimeID和子分区索引重分区后写入
df.repartition(col("TimeID"), getSubPartition(col("TimeID")))
  .write.partitionBy("TimeID")
  .parquet("/path/")

方案3：使用Spark内置参数（简单高效） #

Spark提供了spark.sql.files.maxRecordsPerFile参数，可限制每个文件的最大记录数。你可以根据单条记录的平均大小估算对应记录数（比如单条记录平均1KB，200MB对应204800条）：

// 设置每个文件最多204800条记录（根据实际记录大小调整）
spark.conf.set("spark.sql.files.maxRecordsPerFile", 204800)

df.repartition("TimeID")
  .write.partitionBy("TimeID")
  .parquet("/path/")

这个方案无需修改数据分区逻辑，但依赖记录大小的稳定性，如果记录大小差异较大，文件大小可能会有波动。

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