基于Pandas验证DataFrame列与单元格值匹配SQL定义规则

我有一张SQL表cmf_data，用于存储字段名称和预期数据类型，DDL如下：

CREATE TABLE mydb.cmf_data (
    cmf_data_id int IDENTITY(1,1) NOT NULL,
    cmf_data_field_name varchar(MAX) COLLATE SQL_Latin1_General_CP1_CI_AS NOT NULL,
    cmf_data_field_data_type varchar(MAX) COLLATE SQL_Latin1_General_CP1_CI_AS NOT NULL,
    CONSTRAINT PK__client_m__BC8D4A1B625F6F4F PRIMARY KEY (cmf_data_id)
);

该表的示例查询结果：

还有一份Excel表格，CSV格式示例：

Foo,Baz,Fizz,Buzz
23.5,44.18,'2022-06-18','Hello there',
24.621,7.3,'2023-01-07','How is it going',
149.0,712.0,'2022-02-09','What a nice day',
11.74,101.03,'2021-10-26','Thank you so much'

需求是用cmf_data表的规则约束Excel导入后的Pandas DataFrame：

1. 从数据库读取cmf_data为DataFrame（cmf_data_df）
2. 读取Excel为DataFrame（excel_df）
3. 校验excel_df的列与cmf_data_df中的cmf_data_field_name完全匹配，不允许多列或缺列，否则抛出验证异常
4. 校验excel_df中每个单元格的值符合对应列的定义类型，无法转换则抛出异常

目前已完成数据读取的代码，但不知道如何实现列匹配和行级类型校验：

# read cmf_data_df from DB
import urllib.parse
from sqlalchemy import create_engine
import pandas as pd
import io

params = urllib.parse.quote_plus("Driver={ODBC Driver 17 for SQL Server};"
                                 f"Server={host};"
                                 f"Database={database};"
                                 f"uid={uid};pwd={pwd}")
engine = create_engine("mssql+pyodbc:///?odbc_connect={}".format(params), fast_executemany=True)

query = "SELECT * FROM mydb.cmf_data"
result = engine.execute(query)
data = result.fetchall()
col = list(result.keys())
cmf_data_df = pd.DataFrame(columns=col, data=data)

# read excel -- I'm actually pulling the file from S3
# which is what the 'obj['Body'].read()' is for
data = pd.read_excel(io.BytesIO(obj['Body'].read()), engine="openpyxl")
excel_df = pd.DataFrame(data)

# how to validate that the columns in 'excel_df' match rows definitions in 'cmf_data_df' ?

# how to perform row-level validation on all the types in 'excel_df'?

1. 列匹配校验 #

提取cmf_data中定义的字段列表，和excel_df的列做完全对比，检查是否存在缺失或多余的列：

# 提取cmf_data中定义的字段列表
required_fields = cmf_data_df['cmf_data_field_name'].tolist()
# 获取excel_df的列列表
excel_fields = excel_df.columns.tolist()

# 检查缺失列和多余列
missing_fields = [field for field in required_fields if field not in excel_fields]
extra_fields = [field for field in excel_fields if field not in required_fields]

if missing_fields or extra_fields:
    error_msg = []
    if missing_fields:
        error_msg.append(f"缺失必填列: {', '.join(missing_fields)}")
    if extra_fields:
        error_msg.append(f"存在多余列: {', '.join(extra_fields)}")
    raise ValueError('; '.join(error_msg))

这段代码会对比两个字段列表，找出缺失和多余的列，一旦存在就抛出包含具体信息的异常。

2. 数据类型校验 #

先建立SQL类型到Pandas兼容类型的映射，然后逐列尝试转换，捕获转换失败的单元格并汇总错误：

import numpy as np

# 定义SQL类型到Pandas类型的映射
type_mapping = {
    'float': np.float64,
    'datetime': 'datetime64[ns]',
    'string': 'string'
}

# 构建字段-类型字典
field_type_dict = dict(zip(cmf_data_df['cmf_data_field_name'], cmf_data_df['cmf_data_field_data_type']))

validation_errors = []

# 遍历每个字段进行类型校验
for field in required_fields:
    expected_type = field_type_dict[field]
    pandas_type = type_mapping[expected_type]
    
    try:
        # 尝试转换列类型
        excel_df[field] = excel_df[field].astype(pandas_type)
    except ValueError:
        # 根据不同类型筛选转换失败的行
        if expected_type == 'float':
            mask = pd.to_numeric(excel_df[field], errors='coerce').isna()
        elif expected_type == 'datetime':
            mask = excel_df[field].apply(lambda x: pd.to_datetime(x, errors='coerce')).isna()
        else:
            # string类型一般不会转换失败，可根据需求调整规则
            mask = False
        
        bad_rows = excel_df[mask]
        for idx, row in bad_rows.iterrows():
            validation_errors.append(f"行{idx+1} 列{field}: 值'{row[field]}'无法转换为{expected_type}类型")

if validation_errors:
    raise ValueError('数据类型校验失败:\n' + '\n'.join(validation_errors))

说明：

• 针对不同类型做针对性转换检查，确保每个单元格都符合定义类型
• 收集所有转换失败的单元格信息，统一抛出异常，方便定位问题

import urllib.parse
from sqlalchemy import create_engine
import pandas as pd
import io
import numpy as np

# read cmf_data_df from DB
params = urllib.parse.quote_plus("Driver={ODBC Driver 17 for SQL Server};"
                                 f"Server={host};"
                                 f"Database={database};"
                                 f"uid={uid};pwd={pwd}")
engine = create_engine("mssql+pyodbc:///?odbc_connect={}".format(params), fast_executemany=True)

query = "SELECT * FROM mydb.cmf_data"
result = engine.execute(query)
data = result.fetchall()
col = list(result.keys())
cmf_data_df = pd.DataFrame(columns=col, data=data)

# read excel -- I'm actually pulling the file from S3
# which is what the 'obj['Body'].read()' is for
data = pd.read_excel(io.BytesIO(obj['Body'].read()), engine="openpyxl")
excel_df = pd.DataFrame(data)

# --------------------------
# 1. 列匹配校验
# --------------------------
required_fields = cmf_data_df['cmf_data_field_name'].tolist()
excel_fields = excel_df.columns.tolist()

missing_fields = [field for field in required_fields if field not in excel_fields]
extra_fields = [field for field in excel_fields if field not in required_fields]

if missing_fields or extra_fields:
    error_msg = []
    if missing_fields:
        error_msg.append(f"缺失必填列: {', '.join(missing_fields)}")
    if extra_fields:
        error_msg.append(f"存在多余列: {', '.join(extra_fields)}")
    raise ValueError('; '.join(error_msg))

# --------------------------
# 2. 数据类型校验
# --------------------------
type_mapping = {
    'float': np.float64,
    'datetime': 'datetime64[ns]',
    'string': 'string'
}

field_type_dict = dict(zip(cmf_data_df['cmf_data_field_name'], cmf_data_df['cmf_data_field_data_type']))
validation_errors = []

for field in required_fields:
    expected_type = field_type_dict[field]
    pandas_type = type_mapping[expected_type]
    
    try:
        excel_df[field] = excel_df[field].astype(pandas_type)
    except ValueError:
        if expected_type == 'float':
            mask = pd.to_numeric(excel_df[field], errors='coerce').isna()
        elif expected_type == 'datetime':
            mask = excel_df[field].apply(lambda x: pd.to_datetime(x, errors='coerce')).isna()
        else:
            mask = False
        
        bad_rows = excel_df[mask]
        for idx, row in bad_rows.iterrows():
            validation_errors.append(f"行{idx+1} 列{field}: 值'{row[field]}'无法转换为{expected_type}类型")

if validation_errors:
    raise ValueError('数据类型校验失败:\n' + '\n'.join(validation_errors))

# 校验通过后的后续处理
print("所有校验通过")

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