如何在Python中比对CSV列数据类型与元数据JSON文件定义？

我需要比对CSV文件中所有列的数据类型与元数据JSON文件的定义是否一致。元数据JSON包含CSV所有列的预期数据类型，比如JSON里customer_name定义为CHAR，就要校验CSV中该字段类型是否为CHAR。

用pandas识别CSV列类型后和JSON比对时，经常出现识别偏差：比如JSON中customer_phone定义为CHAR，但pandas会把带前导零的手机号（比如0001234567）识别成INTEGER。想知道Python里有没有更靠谱的实现方法？

示例文件 #

元数据JSON #

{
   "DataType": {
    "customer_name": "CHAR",
    "customer_id": "INTEGER",
    "customer_address": "CHAR",
    "customer_phone": "CHAR"
    }
}

CSV数据 #

customer_name|customer_id|customer_address|customer_phone
abc|12345|abc street|0001234567

我写的函数 #

def check_datatype(df, meta_dict):
    try:
        datatype_issue = False
        file_datatype = df.dtypes.to_dict()
        # 生成要比对的字段列表
        compare_keys = [k for k, v in meta_dict['DataType'].items()]
        
        for key in compare_keys:
            if file_datatype[key] == 'int64':
                file_datatype[key] = 'INTEGER'
            if file_datatype[key] == 'object':
                file_datatype[key] = 'CHAR'
            if file_datatype[key] == 'float64':
                file_datatype[key] = 'DECIMAL'
                
            # 比对类型
            if file_datatype[key] != meta_dict['DataType'][key].lower():
                datatype_issue = True
                
        if not datatype_issue:
            print(f"No datatype issue found")
            
        return datatype_issue
        
    except Exception as e:
        print(e)

方法1：读取CSV时强制指定列类型 #

pandas自动推断类型出错的核心原因是它会根据列内数据的“直观形态”判断类型。解决这个问题的第一步，就是读取CSV时直接根据元数据指定每列的类型，彻底避免自动推断的偏差。

比如根据你的元数据JSON，读取CSV时可以这样做：

import pandas as pd
import json

# 加载元数据
with open('metadata.json', 'r') as f:
    meta_dict = json.load(f)

# 构建元数据类型到pandas类型的映射
type_mapping = {
    'CHAR': str,
    'INTEGER': int,
    'DECIMAL': float
}

# 生成列类型参数：从元数据中提取每个列对应的pandas类型
dtype_spec = {col: type_mapping[meta_dict['DataType'][col]] for col in meta_dict['DataType']}

# 读取CSV，指定分隔符为|，并强制设置列类型
df = pd.read_csv('data.csv', sep='|', dtype=dtype_spec)

这样读取出来的customer_phone会被强制识别为字符串，不会被转成整数。之后再用你原来的校验函数比对，就能得到准确结果。

方法2：逐值校验数据类型（更严谨） #

如果需要更严格的校验（比如确保列内所有值都符合元数据类型，而不是只看pandas的列类型），可以逐行检查每个值的实际类型（或转换可能性）：

def validate_row_types(row, meta_dict):
    errors = []
    for col, expected_type in meta_dict['DataType'].items():
        value = row[col]
        # 处理空值（根据需求调整，比如允许空值则跳过）
        if pd.isna(value):
            continue
            
        # 根据预期类型判断值是否符合
        if expected_type == 'CHAR':
            if not isinstance(value, str):
                errors.append(f"列{col}的值{value}不是CHAR类型")
        elif expected_type == 'INTEGER':
            # 处理CSV中整数以字符串存储的情况
            try:
                int(value)
            except ValueError:
                errors.append(f"列{col}的值{value}无法转换为INTEGER类型")
        elif expected_type == 'DECIMAL':
            try:
                float(value)
            except ValueError:
                errors.append(f"列{col}的值{value}无法转换为DECIMAL类型")
    return errors

# 先把所有列读成字符串，避免pandas自动转换
df = pd.read_csv('data.csv', sep='|', dtype=str)

# 遍历所有行校验
all_errors = []
for idx, row in df.iterrows():
    row_errors = validate_row_types(row, meta_dict)
    if row_errors:
        all_errors.append(f"第{idx+1}行错误: {'; '.join(row_errors)}")

# 输出结果
if all_errors:
    print("发现类型错误：")
    for err in all_errors:
        print(err)
else:
    print("所有列数据类型符合元数据定义")

这种方法完全不依赖pandas的自动类型推断，直接校验每个值的合法性，适用于对数据准确性要求极高的场景。

优化你原有的函数 #

你的函数存在逻辑疏漏（比如大小写比对不一致、判断条件写反），优化后的版本如下：

def check_datatype(df, meta_dict):
    try:
        datatype_issue = False
        # 统一转成大写，避免大小写不一致导致的比对错误
        expected_types = {k: v.upper() for k, v in meta_dict['DataType'].items()}
        file_types = {}
        
        for col in expected_types:
            # 将pandas列类型转换为元数据对应的类型名称
            dtype = df[col].dtype
            if dtype == 'int64':
                file_types[col] = 'INTEGER'
            elif dtype == 'object':
                file_types[col] = 'CHAR'
            elif dtype == 'float64':
                file_types[col] = 'DECIMAL'
            else:
                # 处理未定义的类型
                file_types[col] = str(dtype).upper()
                
            # 比对类型并输出错误信息
            if file_types[col] != expected_types[col]:
                print(f"列{col}类型不匹配：预期{expected_types[col]}，实际{file_types[col]}")
                datatype_issue = True
                
        if not datatype_issue:
            print("未发现类型问题")
            
        return datatype_issue
        
    except Exception as e:
        print(f"校验出错：{e}")

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