如何将含Int64类型的Pandas DataFrame写入MS Access表

环境配置 #

• Python 3.11.8
• SQLAlchemy 2.0.25
• sqlalchemy-access 2.0.2
• pyodbc 5.0.1
• Pandas 2.2.1
• MS Access 2016

连接代码：

engine = sa.create_engine("access+pyodbc://@my_accdb")
conn = engine.connect()

核心问题 #

读取Access表中col1（11位数字，原字段类型为Large Number）和col2（5位数字，原字段类型为长整型Number），将列转为int64后写入新表时触发错误：

sqlalchemy.exc.DBAPIError: (pyodbc.Error) ('HYC00', '[HYC00] [Microsoft][ODBC Microsoft Access Driver]Optional feature not implemented  (106) (SQLBindParameter)')
[SQL: INSERT INTO [new_table] ([Col1], [Col2]) VALUES (?, ?)]
[parameters: [(72001956300, 601), (72001956400, 601), ...]]

转为float64可成功写入，但需保留整数类型；直接使用pyodbc也会触发相同错误。

后续更新情况 #

• 移除astype('int64')步骤后代码可正常运行，但其他函数中出现UnicodeDecodeError: 'utf-16-le' codec can't decode bytes in position 0-1: illegal UTF-16 surrogate
• 怀疑源数据存在问题，导出为CSV时也会出现解码错误

针对int64写入Access的错误 #

1. 调整数据类型映射+事后修改字段类型
   先以字符串类型写入col1，避免ODBC驱动的64位整数绑定限制，写入完成后再将Access字段改回Large Number类型：

   df_dtypes = {
       "col1": sa.types.String,
       "col2": sa.types.Integer
   }
   df.to_sql("new_table", conn, index=False, if_exists='replace', dtype=df_dtypes)
   # 通过SQL语句修改字段类型
   from sqlalchemy import text
   conn.execute(text("ALTER TABLE new_table ALTER COLUMN col1 LONGLONG"))
   conn.commit()
   

2. 使用pyodbc直接批量插入
   绕过Pandas的参数绑定逻辑，手动将整数转为字符串后执行插入：

   cursor = conn.connection.cursor()
   # 先创建对应类型的表
   cursor.execute("CREATE TABLE new_table (col1 LONGLONG, col2 LONG)")
   # 转换参数格式
   params = [(str(row.col1), row.col2) for _, row in df.iterrows()]
   cursor.executemany("INSERT INTO new_table (col1, col2) VALUES (?, ?)", params)
   conn.commit()
   

3. 修改Pandas写入参数
   启用method='multi'，减少参数绑定次数（注意：数据量过大时可能触发Access的SQL语句长度限制）：

   df.to_sql("new_table", conn, index=False, if_exists='replace', dtype=df_dtypes, method='multi')
   

针对Unicode解码错误 #

1. 读取时指定编码
   在连接字符串中指定编码，避免utf-16解码冲突：

   engine = sa.create_engine("access+pyodbc://@my_accdb?charset=utf8")
   

2. 清理源数据中的非法字符
   对字符串列进行代理字符清理，解决解码错误：

   def clean_surrogates(s):
       if isinstance(s, str):
           return s.encode('utf-16', 'surrogatepass').decode('utf-16')
       return s
   
   # 对所有字符串列应用清理逻辑
   for col in tab20.select_dtypes(include=['object']).columns:
       tab20[col] = tab20[col].apply(clean_surrogates)
   

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