如何用PySpark实现多Parquet文件Row Group统计值的有序分布？

要实现你需要的Parquet文件分区与Row Group范围分布，核心思路是自定义范围分区+分区内局部排序，既避免全表orderBy的性能损耗，又能严格控制每个文件、每个Row Group的longColumn范围。以下是具体实现步骤：

1. 预计算全局范围边界 #

首先明确所有目标Row Group的longColumn范围边界。根据需求，64个文件×3个Row Group，共需192个连续区间，每个区间对应一个Row Group的范围。

以你给出的2个文件场景为例，边界列表可定义为：

row_group_bounds = [99, 8663, 8995, 13987, 14334, 19846, 19878, 25622, 25655, 30092, 31095, 35765]

每个Row Group的范围为左闭右开[bounds[i], bounds[i+1})，你可以根据全局数据的min/max和每个Row Group的目标大小（128MB）调整边界，确保每个区间的数据量对应约128MB的存储大小。

2. 为数据标记所属的Row Group与文件ID #

通过条件判断为每条数据添加row_group_id（标记属于哪个Row Group）和file_id（标记属于哪个文件，每3个连续Row Group对应一个文件）：

from pyspark.sql import functions as F

# 生成Row Group归属判断条件
rg_conditions = []
for i in range(len(row_group_bounds) - 1):
    lower = row_group_bounds[i]
    upper = row_group_bounds[i+1]
    # 构建区间判断条件
    cond = (F.col("longColumn") >= lower) & (F.col("longColumn") < upper)
    rg_conditions.append(F.when(cond, i))

# 添加row_group_id列
df_with_rg = df.withColumn("row_group_id", F.coalesce(*rg_conditions))

# 计算文件ID：每3个Row Group合并为一个文件
df_with_file = df_with_rg.withColumn("file_id", F.floor(F.col("row_group_id") / 3))

3. 配置Parquet Row Group大小 #

设置Spark参数，确保每个Row Group的大小约为128MB：

# 设置Row Group大小为128MB（单位：字节）
spark.conf.set("spark.sql.parquet.rowGroupSize", 128 * 1024 * 1024)

4. 分区写入Parquet文件 #

通过repartition按file_id分区（每个分区对应一个输出文件），再通过sortWithinPartitions在分区内按row_group_id和longColumn局部排序，确保每个Row Group的连续范围数据被写入同一个Parquet Row Group：

df_with_file.repartition("file_id") \
            .sortWithinPartitions("row_group_id", "longColumn") \
            .write \
            .mode("overwrite") \
            .parquet("/your/output/path")

性能优势说明 #

• 避免全表orderBy的全局shuffle：仅在每个文件对应的分区内做局部排序，IO和计算开销远低于全表排序。
• 预计算边界仅需一次全局min/max统计，是O(n)的轻量操作。
• repartition("file_id")仅按文件ID做数据shuffle，数据移动量远小于全表排序。

注意事项 #

• 边界校准：如果需要精准控制每个Row Group的大小，可先计算longColumn的直方图，根据数据分布调整边界，确保每个区间的数据量对应128MB左右的存储。
• 空值处理：若longColumn存在空值，需单独添加条件分配到特定Row Group或过滤。
• 压缩格式：可通过spark.sql.parquet.compression.codec设置压缩格式（如snappy），不影响Row Group的范围划分，但会优化文件大小。

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