如何基于pyarrow Schema验证Python对象？

PyArrow的Table.from_pylist默认只会提取Schema中定义的字段，忽略字典里的多余键，不会抛出异常。要实现类似pyarrow.json.ParseOptions.unexpected_field_behavior = "error"的验证效果，可以用以下两种方法：

方法一：手动遍历验证（高效灵活） #

直接提取Schema的字段名集合，逐个检查每个字典的键是否存在多余项：

import pyarrow as pa

# 定义你的Schema
SCHEMA = pa.schema([
    ("id", pa.int64()),
    ("name", pa.string()),
    ("age", pa.int32())
])

def validate_python_objects(obj_list, schema):
    allowed_fields = set(schema.names)
    for index, obj in enumerate(obj_list):
        extra_keys = set(obj.keys()) - allowed_fields
        if extra_keys:
            raise ValueError(f"第{index}个对象包含未定义字段: {', '.join(extra_keys)}")

# 测试用例
sample_objects = [
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25, "extra_field": "invalid"}  # 会触发异常
]

# 先验证再创建表
validate_python_objects(sample_objects, SCHEMA)
table = pa.Table.from_pylist(sample_objects, schema=SCHEMA)

这种方法的优势：

• 无需额外序列化，性能更高，适合处理大数据量
• 可以自定义异常类型和提示信息
• 支持扩展嵌套结构的递归验证（需自行实现嵌套字段检查逻辑）

方法二：借助PyArrow JSON解析器（原生验证逻辑） #

如果可以接受将Python字典序列化为JSON字符串，直接利用PyArrow内置的JSON验证逻辑：

import pyarrow as pa
import json

SCHEMA = pa.schema([
    ("id", pa.int64()),
    ("name", pa.string()),
    ("age", pa.int32())
])

sample_objects = [
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25, "extra_field": "invalid"}
]

# 将对象序列化为JSON字符串
json_data = json.dumps(sample_objects)
# 设置解析选项，遇到未定义字段抛出异常
parse_opts = pa.json.ParseOptions(unexpected_field_behavior="error")

try:
    # 解析时自动验证Schema匹配性
    table = pa.json.read_json(pa.BufferReader(json_data.encode()), schema=SCHEMA, parse_options=parse_opts)
except pa.ArrowInvalid as err:
    print(f"验证失败: {err}")

这种方法的优势：

• 无需手动编写字段检查逻辑，利用PyArrow原生实现
• 自动支持嵌套Schema的验证
• 缺点是需要额外的JSON序列化/反序列化步骤，性能略低

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