Cloud Function+pytds+pandas读取Cloud SQL数据遇UnicodeDecodeError编码错误

针对你用pytds+pandas读取SQL Server数据时遇到的编码错误，以下是几个可行的解决方向：

1. 确认数据库实际编码 #

先排查数据库和目标表的字符集，避免用错编码：

-- 查看数据库默认排序规则（对应编码）
SELECT name, collation_name FROM sys.databases WHERE name = '{db_name}';
-- 查看article表各列的排序规则
SELECT name, collation_name FROM sys.columns WHERE object_id = OBJECT_ID('article');

比如Chinese_PRC_CI_AS对应GBK编码，SQL_Latin1_General_CP1_CI_AS对应Latin1，若数据库用的是UTF8相关排序规则，直接指定encoding="utf-8"更合适。

2. 调整pytds连接参数 #

pytds的bytes_to_unicode和编码设置可能冲突，尝试修改连接逻辑：

def getconn() -> pytds.Connection:
    conn = connector.connect(
        instance_connection_name,
        "pytds",
        user=db_user,
        password=db_pass,
        db=db_name,
        bytes_to_unicode=False,  # 关闭自动转码，手动处理
        encoding="utf-8"  # 替换为数据库实际编码
        )
    return conn

同时移除sqlalchemy create_engine里的connect_args={"encoding": "latin1"}，避免参数冲突。

3. 手动修复字符串列编码 #

如果部分列存在特殊字符，读取后手动处理编码错误：

query = 'select * from article'
df = pd.read_sql(query, pool.connect())

# 遍历所有字符串列，修复编码
for col in df.select_dtypes(include=['object']).columns:
    # 先按原编码转字节，再用目标编码解码，错误字符替换为?
    df[col] = df[col].apply(
        lambda x: x.encode('latin1').decode('utf-8', errors='replace') 
        if isinstance(x, str) else x
    )

4. 排查并清理特殊字符 #

找出表中包含非标准ASCII的行，针对性处理：

-- 替换column_name为怀疑有问题的列名
SELECT * FROM article WHERE column_name LIKE '%[^ -~]%';

可以用REPLACE函数替换特殊字符，或者导出前清理数据。

5. 切换到pyodbc驱动 #

如果pytds的编码问题难以解决，尝试用pyodbc驱动连接：

import pyodbc

def getconn():
    conn_str = (
        f"DRIVER={{ODBC Driver 17 for SQL Server}};"
        f"SERVER={instance_connection_name};"
        f"DATABASE={db_name};"
        f"UID={db_user};"
        f"PWD={db_pass};"
        f"CHARSET=UTF8"
    )
    return pyodbc.connect(conn_str)

pool = sqlalchemy.create_engine(
    "mssql+pyodbc://",
    creator=getconn,
    pool_size=20,
    max_overflow=0,
    pool_timeout=600,
    pool_recycle=1800
)

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