如何用Python拆分多列Excel，解决SQL Server行大小超限问题

错误原因 #

SQL Server的单条记录最大行大小限制为8060字节。你的代码将所有文本类型列映射为NVARCHAR(255)（每个占用510字节），1009列的总大小远超阈值，因此触发报错：

The statement has been terminated. (3621); [42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Cannot create a row of size 8349 which is greater than the allowable maximum row size of 8060

解决方案 #

将原Excel拆分为多个子表，每个子表保留首列作为关联主键，其余列拆分到不同子表中，确保每个子表的行大小不超过8060字节，再分别导入SQL Server。

完整代码 #

# 数据类型映射函数：将Pandas数据类型转换为SQL Server兼容类型
def sqlcol(dfparam):    
    import sqlalchemy as sqla
    dtypedict = {}
    for col_name, dtype in zip(dfparam.columns, dfparam.dtypes):
        dtype_str = str(dtype)
        if "object" in dtype_str:
            dtypedict[col_name] = sqla.types.NVARCHAR(length=255)
        elif "datetime" in dtype_str:
            dtypedict[col_name] = sqla.types.DateTime()
        elif "float" in dtype_str:
            dtypedict[col_name] = sqla.types.Float()
        elif "int" in dtype_str:
            dtypedict[col_name] = sqla.types.BIGINT()
    return dtypedict

# SQL Server数据导入函数
def import_varchar_to_hst03(db: str, tb_name: str, df):
    import sqlalchemy as sqla
    import urllib
    import pyodbc
    dtype_map = sqlcol(df)
    conn_str = urllib.parse.quote_plus(
        f"DRIVER={{ODBC Driver 17 for SQL Server}};SERVER=localhost;DATABASE={db};Trusted_Connection=yes;"
    )
    engine = sqla.create_engine(f'mssql+pyodbc:///?odbc_connect={conn_str}', fast_executemany=True)
    df.to_sql(
        tb_name, 
        schema='dbo', 
        con=engine, 
        index=False,
        dtype=dtype_map,
        if_exists='replace'
    )

# DataFrame拆分函数：按列拆分，保留首列作为主键
def split_dataframe(df, max_cols_per_group=250):
    key_column = df.columns[0]
    key_df = df[[key_column]]
    non_key_columns = df.columns[1:]
    # 计算需要拆分的组数
    num_groups = (len(non_key_columns) + max_cols_per_group - 1) // max_cols_per_group
    split_result = []
    for group_idx in range(num_groups):
        start = group_idx * max_cols_per_group
        end = start + max_cols_per_group
        current_cols = non_key_columns[start:end]
        current_df = pd.concat([key_df, df[current_cols]], axis=1)
        split_result.append((f"{group_idx+1}", current_df))
    return split_result

if __name__ == "__main__":
    import pandas as pd
    
    # 读取目标Excel文件
    excel_path = r'C:\Users\sriram.ramasamy\Desktop\Testsriram.xlsx'
    excel_sheets = pd.read_excel(excel_path, sheet_name=None)
    
    # 遍历每个工作表进行拆分和导入
    for sheet_name, original_df in excel_sheets.items():
        print(f"正在处理工作表: {sheet_name}")
        split_groups = split_dataframe(original_df)
        for group_num, df_group in split_groups:
            target_table = f"{sheet_name}_part_{group_num}"
            print(f"正在导入子表: {target_table}")
            import_varchar_to_hst03('InsightMaster', target_table, df_group)
    
    print('所有数据已成功导入数据库')

关键说明 #

1. 拆分逻辑：

   • 每个子表强制保留原表的首列（作为关联主键），后续查询可通过该列关联所有子表数据
   • max_cols_per_group参数可根据实际数据类型调整：如果文本列占比高，建议调小（比如设为15）；如果数值/日期列多，可适当调大（比如300）

2. 子表命名规则：
   原表名后追加_part_序号，例如原表Sheet1拆分后生成Sheet1_part_1、Sheet1_part_2等，便于识别和关联。

3. 可选扩展：
   如果需要将拆分后的结果保存为新Excel文件，可在拆分后添加以下代码：

   writer = pd.ExcelWriter('拆分后的Excel文件.xlsx')
   for part_name, df_part in split_groups:
       df_part.to_excel(writer, sheet_name=f"{sheet_name}_part_{part_name}", index=False)
   writer.close()
   

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