Pyspark如何获取Nested Struct嵌套结构体列的数据类型

核心思路 #

DataFrame的Schema本质是树状的StructType结构，每一层嵌套的结构体字段本身也是StructType类型，包含多个StructField子节点，我们可以通过路径匹配或者递归遍历的方式逐层定位到目标字段，获取其数据类型。

方案1：单字段路径查询（针对性强） #

适用于已知目标字段完整路径的场景，直接按路径逐层匹配查找：

from pyspark.sql.types import StructType

def get_nested_field_type(df, field_path):
    path_parts = field_path.split(".")
    current_schema = df.schema
    for path_node in path_parts:
        if not isinstance(current_schema, StructType):
            raise ValueError(f"路径节点 {path_node} 所在层级不是结构体，无法继续查找")
        try:
            current_field = current_schema[path_node]
        except KeyError:
            raise ValueError(f"字段路径 {field_path} 中不存在节点 {path_node}")
        current_schema = current_field.dataType
    return current_schema

用法示例 #

要获取层级user.base_info.lastname的字段类型，直接调用：

lastname_type = get_nested_field_type(df, "user.base_info.lastname")
# 可调用simpleString()方法获取可读的类型字符串
print(lastname_type.simpleString())

返回结果示例：string、integer、timestamp等。

方案2：全量嵌套字段打平（适合批量校验） #

如果需要一次性校验所有嵌套字段的类型，可以先将整个Schema打平为「完整字段路径: 数据类型」的映射字典，后续直接查询字典即可，适合多字段批量校验的场景：

def flatten_schema(schema, parent_path=""):
    field_type_map = {}
    for field in schema.fields:
        full_path = f"{parent_path}.{field.name}" if parent_path else field.name
        if isinstance(field.dataType, StructType):
            # 递归处理嵌套结构体
            field_type_map.update(flatten_schema(field.dataType, full_path))
        else:
            field_type_map[full_path] = field.dataType
    return field_type_map

用法示例 #

# 生成全量字段类型映射表
all_field_types = flatten_schema(df.schema)
# 直接查询指定字段类型
print(all_field_types.get("user.base_info.lastname").simpleString())

扩展说明 #

• 如果嵌套结构中包含ArrayType（比如数组元素为结构体的场景），可在上述函数中增加ArrayType判断逻辑，取elementType属性继续递归，即可支持数组嵌套结构体的字段类型查询。

该方案可直接适配多JSON类型校验场景：批量加载每个JSON文件为DataFrame后，调用上述方法获取目标字段类型，提前识别类型不一致的文件，或者统一做类型转换后再写入Delta表，避免写入时的Schema冲突。

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