Spark如何获取嵌套Struct类型列的数据类型

你混淆了PySpark的Column对象和Python原生类型的判断逻辑：col("reportData.ecus.element.ecuId")以及jsonDF.reportData.ecus.element.ecuId返回的都是Spark的Column类实例，不是实际的数据集值。Python的isinstance、type是针对Python对象做类型判断的函数，将这两个函数的返回值传给withColumn的第二个参数（要求必须是Column表达式），自然会触发「col should be Column」的报错。

根据你的使用场景分为两类实现方式：

场景1：获取Schema定义的列类型（元数据校验用） #

如果是要做写入Delta前的列存在性、类型一致性校验，直接解析DataFrame的Schema元数据即可，不需要扫描全量数据，效率最高：

from pyspark.sql.types import StructType, ArrayType
from pyspark.sql.types import IntegerType, StringType, BooleanType

def get_nested_column_type(df, nested_path):
    path_segments = nested_path.split(".")
    current_type = df.schema
    for seg in path_segments:
        # 处理Struct嵌套结构
        if isinstance(current_type, StructType):
            if seg not in current_type.fieldNames():
                raise ValueError(f"嵌套路径不存在，缺失字段：{seg}")
            current_type = current_type[seg].dataType
        # 处理Array数组结构
        elif isinstance(current_type, ArrayType):
            current_type = current_type.elementType
            if seg not in current_type.fieldNames():
                raise ValueError(f"嵌套路径不存在，缺失字段：{seg}")
            current_type = current_type[seg].dataType
        else:
            raise ValueError(f"路径层级错误，{seg} 不是嵌套结构字段")
    return current_type

# 调用示例
try:
    ecu_id_type = get_nested_column_type(df, "reportData.ecus.ecuId")
    # 校验类型是否符合要求，比如要求是字符串类型
    if not isinstance(ecu_id_type, StringType):
        raise TypeError(f"ecuId类型不符合要求，当前类型：{ecu_id_type}")
except (ValueError, TypeError) as e:
    # 校验不通过的逻辑，比如终止任务、记录异常
    print(f"Schema校验失败：{e}")

场景2：逐行获取数据实际类型 #

如果存在同一列不同行的数据类型不一致的情况，需要逐行判断实际值类型，使用Spark内置的typeof函数实现：

from pyspark.sql.functions import typeof, col, when

# 新增列存储每行ecuId的实际类型
df = df.withColumn("ecuId_datatype", typeof(col("reportData.ecus.element.ecuId")))
# 可以结合when函数做类型判断标记
df = df.withColumn(
    "is_ecuId_string",
    when(col("ecuId_datatype") == "string", True).otherwise(False)
)

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