本文详解如何在加载 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() } ) - 生产环境建议：将校验逻辑封装为可复用函数，并集成到 ETL 流程中；对关键字段（如主键、外键）增加非空、唯一性等业务规则校验。
通过以上方法，您可构建高鲁棒性的 CSV 入库前校验机制，显著降低因类型不匹配导致的数据库错误或数据失真风险。
