Spark DataFrame含Decimal列的schema校验忽略精度方案咨询

你需要校验Spark DataFrame的schema与外部来源的字段信息，仅校验字段名和类型大类，Decimal类型忽略精度/标度参数，不校验空值属性。你当前把Decimal转Float再比对的方案有明显缺陷：

• 容易把真实的Float类型和Decimal类型混淆，导致误判
• 无意义的类型转换增加额外开销
• 扩展性差，后续要适配其他带参数的类型时难以修改

直接比对类型的所属大类，不需要转换原始schema，实现如下：

from pyspark.sql import types as T

def is_schema_valid(schema_info: list[tuple[str, str]], df_schema: T.StructType, check_order: bool = True) -> bool:
    # 构建参考schema映射：字段名 -> 期望类型
    expect_schema = {col_name: col_type.lower() for col_name, col_type in schema_info}
    
    # 要求字段顺序严格一致的场景
    if check_order:
        if len(schema_info) != len(df_schema.fields):
            return False
        for (expect_name, expect_type), actual_field in zip(schema_info, df_schema.fields):
            # 校验字段名
            if expect_name != actual_field.name:
                return False
            # 校验类型大类
            actual_type = actual_field.dataType
            if expect_type == "decimal":
                if not isinstance(actual_type, T.DecimalType):
                    return False
            else:
                # 其他类型取简化名比对，比如StringType对应"string"
                actual_type_simple = type(actual_type).__name__.replace("Type", "").lower()
                if actual_type_simple != expect_type:
                    return False
        return True
    # 不要求字段顺序的场景
    else:
        for actual_field in df_schema.fields:
            actual_name = actual_field.name
            if actual_name not in expect_schema:
                return False
            expect_type = expect_schema[actual_name]
            actual_type = actual_field.dataType
            if expect_type == "decimal":
                if not isinstance(actual_type, T.DecimalType):
                    return False
            else:
                actual_type_simple = type(actual_type).__name__.replace("Type", "").lower()
                if actual_type_simple != expect_type:
                    return False
        # 校验是否存在参考schema有、但df缺失的字段
        return set(expect_schema.keys()) == {f.name for f in df_schema.fields}

用你给出的测试用例验证：

df_schema = T.StructType([
    T.StructField('column_1', T.StringType(), True),
    T.StructField('column_2', T.DecimalType(20,5), True),
])

schema_info = [('column_1', 'String'), ('column_2', 'Decimal')]

print(is_schema_valid(schema_info, df_schema))
# 输出: True

• 完全不需要修改原始schema，零额外计算开销
• 不会出现Decimal和Float混淆的误判问题
• 支持顺序校验/非顺序校验两种场景，可按需选择
• 扩展性极强，后续如果要适配其他带参数的类型（比如ArrayType忽略元素类型、StructType忽略内部结构），只要加对应判断规则即可

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