如何在Polars中检查DataFrame的Schema宽松匹配与子集关系？

1. 验证DataFrame数据可宽松匹配指定Schema（类型可向上转换） #

核心是确认DataFrame的每一列类型都能安全向上转换为目标Schema对应列的类型。Polars提供的pl.DataType.can_cast()方法可以直接判断两种类型间的安全转换可能性，基于此实现检查：

• 操作步骤：
  • 分别获取DataFrame的当前Schema和目标Schema
  • 遍历目标Schema的所有列，先检查DataFrame中是否存在该列，再验证该列类型能否安全转换为目标类型
  • 所有列通过检查则返回True，否则返回False

代码示例：

import polars as pl
from collections import OrderedDict

def is_relaxed_schema_match(df: pl.DataFrame, target_schema: OrderedDict) -> bool:
    df_schema = df.schema
    for col_name, target_dtype in target_schema.items():
        if col_name not in df_schema:
            return False
        if not df_schema[col_name].can_cast(target_dtype):
            return False
    return True

# 测试匹配场景
df = pl.DataFrame({"x": [1, 2, 4]})
target_schema = OrderedDict({"x": pl.Float64})
print(is_relaxed_schema_match(df, target_schema))  # 输出: True

# 测试不匹配场景（int无法安全转为字符串）
target_schema_invalid = OrderedDict({"x": pl.Utf8})
print(is_relaxed_schema_match(df, target_schema_invalid))  # 输出: False

2. 验证DataFrame的Schema是另一Schema的子集 #

子集要求：DataFrame的所有列列名和类型必须完全匹配目标Schema中的对应列，目标Schema可包含额外列。

• 操作步骤：
  • 将DataFrame的Schema转换为和目标Schema一致的OrderedDict格式
  • 遍历DataFrame的所有列，检查每一列的列名和类型是否都在目标Schema中存在且完全一致
  • 满足条件则返回True，否则返回False

代码示例：

def is_schema_subset(df: pl.DataFrame, target_schema: OrderedDict) -> bool:
    df_schema = df.schema
    for col_name, dtype in df_schema.items():
        if col_name not in target_schema or target_schema[col_name] != dtype:
            return False
    return True

# 测试子集场景
df = pl.DataFrame({"x": [1, 2, 4]})
target_schema = OrderedDict({"x": pl.Int64, "y": pl.Int64})
print(is_schema_subset(df, target_schema))  # 输出: True

# 测试非子集场景（列类型不匹配）
target_schema_invalid = OrderedDict({"x": pl.Float64, "y": pl.Int64})
print(is_schema_subset(df, target_schema_invalid))  # 输出: False

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