使用pyodbc+SQLAlchemy调用pd.to_sql()写入MySQL时出现编码相关语法错误的问题排查与解决方案咨询

Let's break down what's going on here and fix this issue step by step.

Root Cause: Malformed Connection String #

First, I spotted a critical mistake in how you're building your ODBC connection string. The urllib3.parse.quote_plus function expects a single complete connection string as its first argument, but you're passing multiple separate fragments. This creates invalid connection parameters, which leads to the garbled syntax errors you're seeing (the driver can't interpret your settings correctly, causing string encoding to break during parameter passing).

Step 1: Fix the Connection String #

Rewrite your connection setup to properly construct the ODBC string first, then encode it. Also, match the charset to your database's latin1 encoding and add Unicode-related parameters to help with string conversion:

import urllib.parse  # Use standard urllib.parse instead of urllib3 for this task
from sqlalchemy import create_engine

# Build a single, complete ODBC connection string
conn_str = (
    f"DRIVER={{MySQL ODBC 8.0 ANSI Driver}};"
    f"SERVER={host}:{port};"
    f"DATABASE={db};"
    f"UID={username};"
    f"PWD={password};"
    f"charset=latin1;"  # Match your database's latin1 encoding
    f"UseUnicode=Yes;"
    f"CharacterSet=latin1"
)
# URL-encode the full connection string
params = urllib.parse.quote_plus(conn_str)
db_engine = create_engine(f"mysql+pyodbc:///?odbc_connect={params}")
connection = db_engine.connect()

Try running pd.to_sql() again after this fix—this alone might resolve the issue.

Step 2: Adjust pd.to_sql() Method #

If the connection fix doesn't work, the problem might be how pandas handles parameter passing with pyodbc. Try using the method='multi' parameter, which tells pandas to generate a single bulk insert statement instead of relying on executemany:

maindf.to_sql(
    "mytable",
    connection,
    if_exists="append",
    index=False,
    chunksize=1000,
    method='multi'  # Use bulk insert syntax
)

This reduces the number of round-trips and avoids some of the parameter encoding issues with pyodbc.

Step 3: Manual Bulk Insert (Bypass pandas to_sql) #

If you still hit issues, you can replicate the manual execute() approach you know works, but scale it for bulk data:

from sqlalchemy import text

# Define your insert statement (match your table's columns)
insert_query = text("INSERT INTO mytable (transcript) VALUES (:transcript)")

# Convert DataFrame to a list of dictionaries (matches parameter names)
bulk_data = maindf.to_dict('records')

# Execute in bulk
connection.execute(insert_query, bulk_data)
connection.commit()  # Commit if your connection isn't in autocommit mode

This gives you full control over parameter passing, just like the working execute() calls you mentioned.

Step 4: Switch to a More Compatible Driver #

Since you already confirmed that using SQLAlchemy with mysqlconnector works flawlessly, this is a reliable fallback. Here's how to set that up:

from sqlalchemy import create_engine

# Use mysqlconnector as the driver, with latin1 charset
db_engine = create_engine(
    f"mysql+mysqlconnector://{username}:{password}@{host}:{port}/{db}?charset=latin1"
)
connection = db_engine.connect()

# Run to_sql as normal
maindf.to_sql("mytable", connection, if_exists="append", index=False, chunksize=1000)

This avoids the pyodbc-specific encoding quirks entirely and leverages a driver that's optimized for MySQL.

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