如何基于非标准Spark格式的分区URI过滤Spark DataFrame？

针对Spark中**非标准URI格式（如/{year}/{month}/{day}而非/year={year}/month={month}/day={day}）**的分区数据集，Catalyst无法自动识别分区信息导致过滤效率低下的问题，可通过以下方案实现无需重写数据、注册元数据或重命名目录的高效过滤：

核心思路：自定义分区提取+分区裁剪提示 #

Spark底层的HadoopFsRelation支持手动指定分区列与分区路径结构，我们可以直接操作这层API，让Catalyst识别自定义分区逻辑，从而触发分区裁剪，避免加载无关分区数据。

PySpark实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from org.apache.spark.sql.execution.datasources import HadoopFsRelation, PartitioningAwareFileIndex
from org.apache.spark.sql.execution.datasources.parquet import ParquetFileFormat

# 初始化SparkSession
spark = SparkSession.builder.appName("CustomPartitionFilter").getOrCreate()

# 定义数据本身的Schema（不含分区列）
data_schema = StructType([
    StructField("id", IntegerType(), nullable=False),
    StructField("content", StringType(), nullable=True)
])

# 定义分区列的Schema，顺序要和目录层级对应（year对应第一层目录，month第二层，day第三层）
partition_schema = StructType([
    StructField("year", IntegerType(), nullable=False),
    StructField("month", IntegerType(), nullable=False),
    StructField("day", IntegerType(), nullable=False)
])

# 数据集根路径
base_path = "path/to/your/dataset"

# 创建分区感知文件索引，让Spark按目录层级解析分区值
file_index = PartitioningAwareFileIndex.apply(
    spark._jsparkSession,
    [base_path],
    spark._jsparkSession.sessionState().conf(),
    None,
    None
)

# 关联数据Schema、分区Schema与文件格式，生成支持分区裁剪的Relation
relation = HadoopFsRelation(
    file_index,
    partition_schema,
    data_schema,
    None,
    ParquetFileFormat(),
    spark._jsparkSession.sessionState().conf()
)

# 转换为DataFrame
df = spark.baseRelationToDataFrame(relation)

# 此时过滤分区列会触发自动分区裁剪
filtered_df = df.filter((df.year == 2024) & (df.month == 5) & (df.day == 1))
filtered_df.show()

关键原理说明 #

1. 手动指定分区Schema：直接告知Spark分区列的名称、类型与顺序，替代默认的键值对格式自动识别逻辑
2. PartitioningAwareFileIndex：让Spark扫描路径时，按目录层级依次映射到分区列（根路径下第一级目录对应第一个分区列，以此类推）
3. HadoopFsRelation：将数据文件、分区规则、文件格式绑定，生成能被Catalyst识别的可裁剪数据源

验证分区裁剪生效的方式 #

故意破坏一个无需读取的分区（比如删除该分区下的部分数据文件），执行过滤操作后如果未报错，说明Spark没有加载该分区的数据，分区裁剪已生效。

内容的提问来源于stack exchange，提问作者Martín Baigorria Alonso