未指定Schema时，如何用Hive风格分区解析路径生成过滤表达式？

报错原因说明 #

当仅指定flavor="hive"不传入schema时，ds.partitioning()返回的是PartitioningFactory对象——这是一个用于自动推断分区Schema的工厂类，而非可直接解析路径的Partitioning实例，因此调用.parse()会触发AttributeError。

无Schema时Hive分区的发现逻辑 #

PyArrow的Hive分区工厂类通过以下方式自动发现分区：

• 扫描目标路径中的Hive风格分区目录（格式为key=value，多级分区用/分隔），提取所有键值对作为分区列。
• 根据分区值的字符串内容自动推断数据类型：比如纯数字字符串转为整数/浮点数，true/false转为布尔型，其他字符串保留为字符串类型。
• 推断过程必须基于实际存在的路径集合完成，无法凭空生成Schema。

生成合并过滤条件的两种方案 #

方案1：利用PyArrow自动推断分区Schema后解析 #

先通过路径集合让工厂类完成Schema推断，得到可用的Partitioning实例，再解析路径生成过滤条件：

import pyarrow.dataset as ds
from pyarrow import fs

# 从Airflow XCom获取的路径列表
paths = ["s3://bucket/path/date_id=20240101/test=1", "s3://bucket/path/date_id=20240102/test=0"]

# 初始化文件系统（本地用fs.LocalFileSystem()，对象存储根据实际类型选择）
file_system = fs.S3FileSystem()

# 创建分区工厂并通过路径推断Schema
partitioning_factory = ds.partitioning(flavor="hive")
partitioning = partitioning_factory.discover(file_system, paths)

# 解析每个路径生成过滤表达式，再合并为OR条件
filters = [partitioning.parse(path) for path in paths]
combined_filter = ds.expression.or_(*filters)

方案2：手动解析路径生成过滤条件（不依赖类型推断） #

如果完全不关心分区列的数据类型，可直接按字符串匹配生成过滤条件，无需依赖PyArrow的推断逻辑：

import re
from pyarrow.dataset import expression as expr

def parse_hive_partition(path):
    # 匹配路径中的所有key=value格式分区
    partition_pattern = r'(\w+)=([^/]+)'
    partition_kv = re.findall(partition_pattern, path)
    # 生成每个分区列的相等条件，用AND连接单路径的多分区
    return expr.and_(*[expr.field(key) == value for key, value in partition_kv])

# 从Airflow XCom获取的路径列表
paths = ["s3://bucket/path/date_id=20240101/test=1", "s3://bucket/path/date_id=20240102/test=0"]

# 生成单路径过滤条件后合并为OR条件
single_path_filters = [parse_hive_partition(p) for p in paths]
combined_filter = expr.or_(*single_path_filters)

方案选择建议 #

• 若需要分区列的正确数据类型（比如数值类型用于范围过滤），优先选择方案1。
• 若仅需匹配分区值、不关心类型，方案2更轻量高效。

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