基于常量列过滤PySpark分区Parquet表的最优方法咨询

你的核心问题在于之前的groupBy和dropDuplicates会触发全表扫描——即便分区数量少，但每个分区数据量极大时，依然会耗时很久。而利用"每个分区内cons_col是常量"这个特性，我们可以只读取每个分区的极小数据（比如一行）来获取cons_col的值，快速筛选出符合条件的分区后，再加载这些分区的全量数据，彻底避免无效的全分区扫描。

具体实现步骤（以Spark为例） #

1. 快速获取每个分区的cons_col值

因为每个分区内cons_col是常量，我们只需要读取每个分区的任意一行数据就能拿到该分区的cons_col值，不需要扫描整个分区：

# 从Hive元数据获取所有分区信息（如果表在Hive中）
partition_rows = spark.sql("SHOW PARTITIONS your_table_name").collect()
part_col_values = [row["partition"].split("=")[1] for row in partition_rows]

# 遍历每个分区，仅读取一行数据获取cons_col值
partition_cons_map = {}
for part_val in part_col_values:
    # 直接指定分区路径读取，避免全表扫描
    cons_val = (
        spark.read.format("parquet")
        .option("path", f"/path/to/your/table/part_col={part_val}")
        .load()
        .select("cons_col")
        .limit(1)
        .first()["cons_col"]
    )
    partition_cons_map[part_val] = cons_val

2. 筛选出符合条件的分区

根据cons_col = 1的条件，筛选出对应的part_col值：

target_partitions = [p for p, cons in partition_cons_map.items() if cons == 1]

3. 加载目标分区的全量数据

只读取筛选后的分区，避免扫描无关分区：

# 直接通过分区条件过滤，Spark会自动触发分区裁剪
filtered_df = (
    spark.read.format("parquet")
    .option("path", "/path/to/your/table")
    .where(f"part_col IN ({','.join(map(str, target_partitions))})")
    .load()
)

为什么这个方案更快？ #

• 第一步仅读取每个分区的一行数据，30个分区的情况下几乎瞬间完成，完全避免了全分区扫描；
• 第三步只加载符合条件的分区数据，无关分区的数据根本不会被读取，大幅减少了IO开销；
• 相比groupBy和dropDuplicates，彻底跳过了对海量数据的 shuffle 和去重操作，性能提升数量级。

额外优化点 #

• 如果你的表是通过Spark DataFrame创建的，也可以用spark.read.parquet("/path/to/table").select("part_col").distinct()先获取所有分区值，再逐个读取分区的一行数据，效果类似；
• 若分区路径有特殊格式（比如字符串类型带引号），注意在拼接IN条件时做转义处理。

内容的提问来源于stack exchange，提问作者Rodrigo Serna Pérez