如何基于列数在Pandas中动态拆分DataFrame适配SQL插入？

问题背景 #

你遇到的SQL Server错误是因为单行总字节数超过了8060的限制，手动指定索引范围的方式无法适配不同列数的Excel文件，需要实现按250列动态拆分且保留首列的逻辑。

完整解决方案代码 #

import pandas as pd
import numpy as np
import sqlalchemy as sqla
import urllib
import pyodbc

def sqlcol(dfparam):    
    dtypedict = {}
    for col_name, dtype in zip(dfparam.columns, dfparam.dtypes):
        if "object" in str(dtype):
            dtypedict[col_name] = sqla.types.NVARCHAR(length=255)
        elif "datetime" in str(dtype):
            dtypedict[col_name] = sqla.types.DateTime()
        elif "float" in str(dtype):
            dtypedict[col_name] = sqla.types.Float()
        elif "int" in str(dtype):
            dtypedict[col_name] = sqla.types.BIGINT()
        elif "decimal" in str(dtype):
            dtypedict[col_name] = sqla.types.DECIMAL()
    return dtypedict

def split_dataframe(df, max_cols_per_part=250):
    """
    动态拆分DataFrame，每个拆分后的DF包含首列+最多max_cols_per_part-1列
    :param df: 原DataFrame
    :param max_cols_per_part: 每个拆分后的DF总列数（含首列）
    :return: 拆分后的DF列表
    """
    split_dfs = []
    total_cols = len(df.columns)
    # 首列索引
    first_col_idx = [0]
    
    # 计算需要拆分的次数：除首列外，每次取max_cols_per_part-1列
    cols_per_split = max_cols_per_part - 1
    # 生成拆分的列段起始索引
    split_starts = range(1, total_cols, cols_per_split)
    
    for start in split_starts:
        end = start + cols_per_split
        # 合并首列和当前列段的索引，避免越界
        col_indices = first_col_idx + list(range(start, min(end, total_cols)))
        split_df = df.iloc[:, col_indices]
        split_dfs.append(split_df)
    
    return split_dfs

def import_varchar_to_hst03(db: str, tb_name: str, split_dfs):
    """
    将拆分后的DF列表导入SQL Server，自动生成表名后缀
    """
    quoted = urllib.parse.quote_plus(
        "DRIVER={ODBC Driver 17 for SQL Server};SERVER=localhost;DATABASE="+db+";Trusted_Connection=yes;"
    )
    engine = sqla.create_engine(
        'mssql+pyodbc:///?odbc_connect={}'.format(quoted), 
        fast_executemany=True
    )
    
    # 第一个表用原表名，后续表加_partN后缀
    for idx, df_part in enumerate(split_dfs):
        target_tb = tb_name if idx == 0 else f"{tb_name}_part{idx}"
        
        dtype_dict = sqlcol(df_part)
        df_part.to_sql(
            target_tb, 
            schema='dbo', 
            con=engine, 
            index=False,
            dtype=dtype_dict,
            if_exists='replace'
        )
        print(f"表 {target_tb} 导入完成")

# 主执行逻辑
if __name__ == "__main__":
    # 读取Excel所有工作表
    excel_path = r'C:\Users\sriram.ramasamy\Desktop\Testsriram.xlsx'
    Ex = pd.read_excel(excel_path, sheet_name=None)
    
    for sheet_name, df in Ex.items():
        print(f"开始处理工作表 {sheet_name}")
        # 动态拆分DF
        split_dfs = split_dataframe(df, max_cols_per_part=250)
        # 导入SQL Server
        import_varchar_to_hst03('InsightMaster', sheet_name, split_dfs)
    
    print('所有数据已成功导入数据库')

关键逻辑说明 #

1. 动态拆分函数split_dataframe

   • 自动计算拆分次数：根据原DF总列数，按“首列+249列”的规则拆分，彻底避免索引越界问题
   • 每次拆分强制保留首列，确保后续SQL关联时能精准匹配行数据
   • 用min(end, total_cols)处理最后一段不足249列的情况，杜绝索引溢出错误

2. 导入函数import_varchar_to_hst03

   • 接收拆分后的DF列表，循环处理每个子DF
   • 自动生成表名：第一个表沿用原工作表名，后续表添加_part1、_part2等后缀，便于识别关联
   • 修复了原代码中引用外部变量的bug，每个子DF单独生成对应的SQL类型映射字典

3. 类型映射函数sqlcol

   • 保留原有类型转换逻辑，将Pandas数据类型精准映射为SQL Server兼容类型，避免插入时的类型不匹配错误

使用说明 #

• 修改excel_path为你的Excel文件实际路径
• 调整max_cols_per_part参数可自定义每个拆分后的DF列数（默认250）
• 导入完成后，所有拆分表都包含首列，可通过首列值在SQL中执行关联查询

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