使用pyodbc操作Azure SQL Server遇Function sequence error的优化咨询

错误原因 #

ODBC HY010 函数序列错误的核心原因是：同一个数据库连接上，活跃的读取游标未完成数据读取时执行了commit操作。commit会终止当前事务，而读取游标依赖当前事务的上下文，导致后续fetchmany调用失去有效事务环境，触发错误。

优化方案 #

1. 采用双连接隔离读写操作 #

创建两个独立的数据库连接：一个专门用于读取数据，另一个专门用于写入和提交事务。这样读取游标的事务不会被写入操作的commit中断，彻底避免函数序列错误。

2. 批量写入替代逐行操作 #

当前代码用apply逐行调用savefunc插入数据，10万行数据会产生10万次SQL执行，性能极差。改用批量插入方式，可将性能提升数十倍甚至上百倍：

• 使用pandas的to_sql方法，通过method='multi'实现批量插入
• 或构造批量INSERT语句，一次插入多行数据

3. 增大读取批次（可选） #

将FETCH_SIZE从5调整为1000或更大（根据内存情况），减少读取数据库的次数，提升读取效率。

修改后的代码示例 #

import pyodbc
import pandas as pd
import logging

# 初始化两个独立连接：一个读，一个写
conn_read = pyodbc.connect("你的读取连接字符串")
conn_write = pyodbc.connect("你的写入连接字符串")

def getStgDataFromDB(conn, dbtbl, step, refloaddatetime, countyfld, countyval):
    if step not in ["clean", "match"]: 
        logging.info(f"Invalid step value: {step}")
        return None

    readCursor = conn.cursor()
    sqlstmt = f"""
        SELECT * FROM {dbtbl} 
        WHERE loaddatetime = CAST(? AS DATETIME) 
        AND LOWER({countyfld}) = ? 
        ORDER BY loaddatetime, {countyfld}
    """
    readCursor.execute(sqlstmt, refloaddatetime, countyval.lower())
    return readCursor

# 配置参数
srctbl = "源表名"
step = "clean"
refloaddatetime = "你的加载时间"
countyfld = "区县字段名"
countyval = "区县值"
FETCH_SIZE = 1000  # 增大读取批次
TARGET_TABLE = "目标表名"

cursor_output = getStgDataFromDB(conn_read, srctbl, step, refloaddatetime, countyfld, countyval)
if not cursor_output:
    logging.error("读取游标初始化失败")
    exit()

# 只获取一次列名，避免重复操作
cols = [desc[0] for desc in cursor_output.description]

tblRows = cursor_output.fetchmany(FETCH_SIZE)
while tblRows:
    # 转换为DataFrame
    tblDataStg = pd.DataFrame(tblRows, columns=cols)
    
    # ----------------------
    # 这里执行你的数据清洗逻辑，得到tblDataClean
    # tblDataClean = ...
    # ----------------------
    
    # 批量写入目标表
    tblDataClean.to_sql(
        name=TARGET_TABLE,
        con=conn_write,
        if_exists='append',
        index=False,
        method='multi',  # 启用批量插入
        chunksize=1000   # 每次批量插入1000行
    )
    
    # 提交当前批次的写入事务
    conn_write.commit()
    
    # 读取下一批数据
    tblRows = cursor_output.fetchmany(FETCH_SIZE)

# 清理资源
cursor_output.close()
conn_read.close()
conn_write.close()

额外说明 #

• 双连接方案是最稳妥的解决方式，完全隔离读写事务，不会出现函数序列错误
• 批量插入是提升性能的关键，相比逐行插入，能大幅减少数据库交互次数
• 若因环境限制只能用单连接，可尝试在连接字符串中开启MARS（MARS_Connection=Yes），允许同一连接上同时存在活跃的读取游标和写入操作，但MARS可能存在性能和兼容性问题，需谨慎测试

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