Spark按列重分区后Parquet文件过大，如何限制单文件至128MB？

针对你按指定列重分区后生成超大Parquet文件的问题，以下是几种无需固定分区数、自动将文件大小控制在128MB左右的方案，适配Dataproc+GCS的环境：

方案一：开启Spark自适应执行计划（推荐） #

Spark的自适应查询执行（AQE）可以根据实际数据量自动调整分区大小，完美适配数据量小时波动的场景。只需配置几个参数，就能让Spark自动将大分区拆分、小分区合并到目标大小：

// Scala示例
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", 128 * 1024 * 1024) // 128MB

df.repartition(col("column_name")).write.parquet("gs://path_of_bucket")

# Python示例
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", 128 * 1024 * 1024)

df.repartition(col("column_name")).write.parquet("gs://path_of_bucket")

原理：AQE会在作业执行过程中实时统计分区数据量，自动将超过128MB的分区拆分为多个符合大小的子分区，同时合并过小的分区，无需手动计算分区数。

方案二：按记录数限制文件大小 #

如果不想依赖AQE，可以通过预估单条记录的平均大小，计算出128MB对应的记录数，然后设置spark.sql.files.maxRecordsPerFile参数，强制每个文件的记录数不超过阈值：

# 假设单条记录平均大小为1KB，128MB约对应131072条记录（根据实际数据调整）
spark.conf.set("spark.sql.files.maxRecordsPerFile", 131072)

df.repartition(col("column_name")).write.parquet("gs://path_of_bucket")

注意：这种方式依赖对单条记录大小的预估，如果数据记录大小波动较大，可能会出现文件大小偏差，适合数据格式相对固定的场景。

方案三：动态拆分大数据量的列分区 #

如果某个column_name的值对应的数据量远大于128MB，可以先统计每个列值的数据量，动态为大数据量的列值添加子分区标识，实现拆分：

from pyspark.sql.functions import col, when, rand, hash

# 预估单条记录大小（单位：字节），根据实际数据调整
size_per_record = 1024
target_size = 128 * 1024 * 1024  # 128MB

# 统计每个column_name值的记录数
group_counts = df.groupBy("column_name").count().collect()

# 生成动态分区规则：大数据量的列值拆分多个子分区
repartition_cols = [col("column_name")]
for row in group_counts:
    col_val = row["column_name"]
    count = row["count"]
    # 计算需要拆分的子分区数
    num_sub_part = max(1, int(count * size_per_record / target_size) + 1)
    # 为该列值添加子分区标识
    repartition_cols.append(when(col("column_name") == col_val, hash(rand()) % num_sub_part).otherwise(0))

# 按组合规则重分区后写入
df.repartition(*repartition_cols).write.parquet("gs://path_of_bucket")

原理：通过预统计确定需要拆分的列值，用随机哈希生成子分区标识，让大数据量的列值被拆分为多个小分区，保证每个输出文件大小接近128MB。

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