pandas to_sql写入SQL Server报列名必须唯一错误的通用解决方法

触发该异常的核心原因有两点：

• pandas读取Excel、CSV文件时会完整保留列名首尾的空白字符（空格、制表符、换行符等），语义完全一致的列仅因为首尾多了空格就会被识别为独立字段
• pd.concat 拼接多文件DataFrame时不会自动做列名清洗、对齐，最终拼接结果中会出现仅首尾空白有差异的重名列，再叠加SQL Server对标识符尾部空格不敏感的规则，就会触发重复列建表错误；同时重名列往往伴随类型不一致的问题（比如你遇到的同名列一个是FLOAT数值型、一个是VARCHAR字符型），即使绕过重复列校验，后续写入也会出现类型不匹配的报错。

不需要针对单个异常文件写特殊适配逻辑，只需要在单个文件读取完成后、加入拼接列表前增加统一的列清洗、重名列合并步骤，最后在拼接阶段做全局类型对齐即可覆盖所有同类场景，可直接复用以下改造后的代码：

import pandas as pd
import numpy as np
from sqlalchemy import create_engine

def clean_and_merge_duplicate_cols(df: pd.DataFrame) -> pd.DataFrame:
    # 清洗列名：去首尾所有空白字符，列名中间连续多空格合并为单个空格
    df.columns = (
        df.columns
        .astype(str)
        .str.strip()
        .str.replace(r'\s+', ' ', regex=True)
    )

    # 合并同文件内的重名列
    if df.columns.duplicated().any():
        merged_data = {}
        for col in df.columns.unique():
            col_subset = df.loc[:, df.columns == col]
            # 优先提取数值类型有效值，兼容指标列被误读为字符串的场景
            numeric_series = pd.to_numeric(col_subset.stack(), errors='coerce').unstack()
            merged_col = numeric_series.bfill(axis=1).iloc[:, 0]
            # 数值列全空时回退取字符列的非空值
            if merged_col.isna().all():
                merged_col = col_subset.bfill(axis=1).iloc[:, 0]
            merged_data[col] = merged_col
        df = pd.DataFrame(merged_data)
    return df

# 改造后的文件读取逻辑
all_df_list = []
for file in target_file_list: # 替换为实际的文件遍历逻辑
    suffix = file.suffix.lower()
    if suffix == '.xlsx':
        frame = pd.read_excel(file, header=0, engine='openpyxl')
        # 此处保留原有的filename、load_date等自定义字段添加逻辑
        # (...)
    elif suffix == '.csv':
        # 自动兼容utf-8、gbk编码，避免个别csv编码不匹配读失败
        try:
            frame = pd.read_csv(file, encoding="utf-8")
        except UnicodeDecodeError:
            frame = pd.read_csv(file, encoding="gbk")
        # 此处保留原有的filename、load_date等自定义字段添加逻辑
        # (...)
    else:
        continue
    # 每个文件入拼接列表前先做列清洗、重名列合并
    cleaned_frame = clean_and_merge_duplicate_cols(frame)
    all_df_list.append(cleaned_frame)

# 拼接时自动对齐所有列
xls = pd.concat(all_df_list, ignore_index=True, sort=False)

# 全局统一字段类型
for col in xls.columns:
    test_num = pd.to_numeric(xls[col], errors='coerce')
    # 非空值80%以上可转数值则统一为FLOAT类型，避免脏数据导致整列变为字符串
    if len(test_num.dropna()) / max(len(xls[col].dropna()), 1) > 0.8:
        xls[col] = test_num
    else:
        xls[col] = xls[col].astype(str).replace('nan', np.nan)

# 按原有逻辑写入数据库
xls.to_sql(table, con=engine, if_exists='append', index=False, chunksize=10000)

方案说明 #

• 列清洗逻辑不会修改列名本身的语义，除了首尾去空、合并连续多空格外不改动列名文本，不会出现清洗后列名和原文件无法对应的问题
• 重名列合并逻辑优先保留数值，适配指标列被空值、格式错误内容误识别为字符串的场景，不会丢失有效数据；如果有自定义的重名合并规则，直接修改clean_and_merge_duplicate_cols函数内的合并逻辑即可
• 全局类型校准的数值判定阈值可根据业务场景自行调整，避免个别行的脏数据导致整列类型和数据库已有字段不匹配
• 整个处理逻辑对所有读取的文件生效，后续遇到任何形式的列名空白、重名、类型混杂问题都不需要单独写适配代码，适合大批量文件处理场景。

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