如何使用 Pandas 验证 CSV 文件列名及对应数据类型是否符合预期-Python教程-PHP中文网

如何使用 Pandas 验证 CSV 文件列名及对应数据类型是否符合预期

本文详解如何在加载 CSV 数据前，准确校验列名是否存在且数据类型匹配预期（注意：Pandas 中字符串列为 object 类型，非 str），并提供健壮的验证函数、类型映射建议及数据库写入时的类型控制技巧。

本文详解如何在加载 csv 数据前，准确校验列名是否存在且数据类型匹配预期（注意：pandas 中字符串列为 `object` 类型，非 `str`），并提供健壮的验证函数、类型映射建议及数据库写入时的类型控制技巧。

在将 CSV 数据导入数据库前进行结构化校验，是确保数据质量与下游系统稳定的关键步骤。但直接用 assert df.dtypes.to_dict() == {"Name": str, "Age": int} 会失败——因为 Pandas 中文本列默认为 object 类型（而非 Python 原生 str），数值列也常为 int64/float64 等具体 NumPy 类型，而非抽象的 int 或 float。

✅ 正确的类型校验方式

应使用 pd.api.types.is_string_dtype()、pd.api.types.is_integer_dtype() 等语义化检查函数，或显式比对 dtype 的字符串表示：

import pandas as pd
from pandas.api import types

def validate_csv_schema(df: pd.DataFrame, expected_schema: dict) -> bool:
    """
    校验 DataFrame 列名及数据类型是否符合预期 schema
    expected_schema: {"列名": "expected_dtype"}，支持 'string', 'integer', 'float', 'boolean', 'datetime'
    """
    # 检查列名存在性
    missing_cols = set(expected_schema.keys()) - set(df.columns)
    if missing_cols:
        raise ValueError(f"缺失必需列: {missing_cols}")

    # 检查每列数据类型
    for col, expected_type in expected_schema.items():
        actual_dtype = df[col].dtype
        if expected_type == "string":
            if not types.is_string_dtype(actual_dtype) and not types.is_object_dtype(actual_dtype):
                raise TypeError(f"列 '{col}' 期望 string 类型，实际为 {actual_dtype}")
        elif expected_type == "integer":
            if not types.is_integer_dtype(actual_dtype):
                raise TypeError(f"列 '{col}' 期望 integer 类型，实际为 {actual_dtype}")
        elif expected_type == "float":
            if not types.is_float_dtype(actual_dtype):
                raise TypeError(f"列 '{col}' 期望 float 类型，实际为 {actual_dtype}")
        elif expected_type == "boolean":
            if not types.is_bool_dtype(actual_dtype):
                raise TypeError(f"列 '{col}' 期望 boolean 类型，实际为 {actual_dtype}")
        else:
            raise ValueError(f"不支持的类型: {expected_type}")
    return True

# 示例使用
data = [['tom', 10], ['nick', 15], ['juli', 14]]
df = pd.DataFrame(data, columns=['Name', 'Age'])
df['Age'] = df['Age'].astype('int64')  # 显式转为整数

try:
    validate_csv_schema(df, {"Name": "string", "Age": "integer"})
    print("✅ Schema 校验通过")
except (ValueError, TypeError) as e:
    print(f"❌ 校验失败: {e}")

⚠️ 注意事项与最佳实践

避免直接比较 dtype == str：Pandas 字符串列始终是 object 类型（即使内容全为字符串），应使用 is_string_dtype() 或 df[col].apply(type).nunique() == 1 辅助判断。
处理空值与混合类型：含缺失值的整数列在 Pandas 中会自动转为 float64；如需保留整数语义，可使用 Int64（nullable integer）扩展类型：df['Age'] = df['Age'].astype("Int64")。

对接数据库写入：使用 to_sql(dtype=...) 显式指定 SQL 类型，避免 Pandas 自动推断偏差：

from sqlalchemy.types import String, Integer, DateTime
df.to_sql(
    "users",
    con=engine,
    if_exists="replace",
    index=False,
    dtype={
        "Name": String(50),
        "Age": Integer()
    }
)