如何在PySpark中校验DataFrame的Schema（列数量与列名是否合规）

核心校验规则 #

• 读取CSV文件生成DataFrame后需满足两个条件：
  • 总列数不少于3
  • 必须完全匹配包含id、name、phone三个列，额外存在的列不受限制
• 不满足规则时程序直接抛出错误

代码实现（Python+pandas） #

import pandas as pd

def validate_csv(csv_path: str) -> pd.DataFrame:
    # 读取CSV，统一转为字符串格式避免ID、手机号等内容格式丢失
    df = pd.read_csv(csv_path, dtype=str)
    required_cols = {"id", "name", "phone"}
    current_cols = set(df.columns)
    
    # 校验列数要求
    if len(df.columns) < 3:
        raise ValueError(f"校验失败：当前文件仅{len(df.columns)}列，要求最少3列")
    # 校验必填列是否齐全
    if not required_cols.issubset(current_cols):
        miss_cols = required_cols - current_cols
        raise ValueError(f"校验失败：缺失必填列{', '.join(miss_cols)}，必须包含id、name、phone三个列")
    
    return df

# 调用示例
if __name__ == "__main__":
    try:
        result_df = validate_csv("待校验的文件路径.csv")
        print("列校验通过，可继续后续数据处理")
    except ValueError as err:
        print(err)

注意：列名校验为完全匹配，大小写敏感，例如ID、Phone不会被识别为对应必填列。

示例说明 #

符合要求的场景 #

以下两种情况均可以通过校验：

+-----+-----+-----+   +-----+-----+-----+-----+
|   id| name|phone|   |   id| name|phone|unit |
+-----+-----+-----+   +-----+-----+-----+-----+
|3940A|jhon |1345 |   |3940A|jhon |1345 | 222 |
|2BB56|mike | 492 |   |2BB56|mike | 492 | 333 |
|3(401|jose |2938 |   |3(401|jose |2938 | 444 |
+-----+-----+-----+   +-----+-----+-----+-----+

不符合要求的场景 #

以下两种情况会触发报错：

+-----+-----+-----+   +-----+-----+
|  sku| nomb|phone|   |  sku| name|
+-----+-----+-----+   +-----+-----+
|3940A|jhon |1345 |   |3940A|jhon |
|2BB56|mike | 492 |   |2BB56|mike |
|3(401|jose |2938 |   |3(401|jose |
+-----+-----+-----+   +-----+-----+

前者缺失id、name两个必填列，后者总列数仅2列且缺失id、phone两个必填列。

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