如何在千列级DataFrame中校验指定Schema字段是否存在

断言失败是因为判断逻辑的类型完全不匹配：

• df.columns 的返回值是字符串列表，每个元素是单列的列名，例如 ["user_id", "column 1", ...]
• 你代码里定义的schema是StructType类的实例，属于Spark SQL的结构体对象，不是字符串类型。拿结构体对象去判断是否属于字符串列表，结果永远为False，必然触发断言错误。

另外你贴的代码里StructType的构造还漏了右闭合括号，本身存在语法问题。

你的需求只是校验自定义的4个字段存在于DataFrame中，不需要把整个StructType对象放到in判断里，参考以下修复后的测试代码：

from pyspark.sql.types import StructType, StructField, StringType

def test_column_names(self):
    # 定义需要校验的目标字段规则
    check_schema = StructType([
        StructField("column 1", StringType(), True),
        StructField("column 2", StringType(), True),
        StructField("column 3", StringType(), True),
        StructField("column 4", StringType(), True)
    ])
    # 提取所有待校验的列名
    required_cols = [field.name for field in check_schema.fields]
    # 逐个校验列存在，断言失败时会明确提示缺失的列名
    for col in required_cols:
        assert col in df.columns, f"DataFrame缺失必要字段：{col}"

• 这段校验逻辑不会受DataFrame其余上千个列是否全为null、是否无有效值的影响，仅检查你指定的4个列是否存在
• 如果后续需要同时校验列的数据类型、是否允许空值等Schema属性，可以在循环中取出DataFrame对应列的元数据，和你定义的StructField的dataType、nullable属性做比对即可。

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