使用SQLAlchemy to_sql上传数据致数据库锁定，求非分批解决方案

问题场景 #

以下是用于将DataFrame上传至SQL Server的函数：

def uploadtoDatabase(data,columns,tablename):

    connect_string = urllib.parse.quote_plus(f'DRIVER={{ODBC Driver 17 for SQL Server}};Server=ServerName;Database=DatabaseName;Encrypt=yes;Trusted_Connection=yes;TrustServerCertificate=yes')
    engine = sqlalchemy.create_engine(f'mssql+pyodbc:///?odbc_connect={connect_string}', fast_executemany=False)

    # define data to upload and chunksize (max 2100 parameters)
    requireddata = columns
    chunksize = 1000//len(requireddata)
    
    #convert boolean columns
    boolcolumns=data.select_dtypes(include=['bool']).columns
    data.loc[:,boolcolumns] = data[boolcolumns].astype(int)
    
    #convert objects to string
    objectcolumns=data.select_dtypes(include=['object']).columns
    data.loc[:,objectcolumns] = data[objectcolumns].astype(str)

    #load data
    with engine.connect() as connection:
      data[requireddata].to_sql(tablename, connection, index=False, if_exists='replace',schema = 'dbo')
      connection.close()

    engine.dispose()

执行该函数上传400万条记录时，数据库会被锁定，其他进程需等待上传完成。除了循环分批上传数据外，可通过以下方案确保上传期间其他数据库事务正常执行：

可行解决方案 #

1. 调整事务隔离级别 #

SQL Server默认的READ COMMITTED隔离级别会在批量写入期间持有锁至事务结束，可切换到快照类隔离级别，让其他事务读取数据快照而非等待锁释放：

• 先在数据库端开启快照支持：

ALTER DATABASE YourDatabaseName SET ALLOW_SNAPSHOT_ISOLATION ON;
ALTER DATABASE YourDatabaseName SET READ_COMMITTED_SNAPSHOT ON;

• 在代码中设置连接的隔离级别：

with engine.connect() as connection:
    connection.exec_driver_sql("SET TRANSACTION ISOLATION LEVEL READ COMMITTED SNAPSHOT")
    data[requireddata].to_sql(tablename, connection, index=False, if_exists='replace',schema = 'dbo')

2. 临时表+原子替换原表 #

当前使用if_exists='replace'会先删除原表再重建，全程持有表级锁。改用临时表上传后再原子替换，可大幅缩短锁持有时间：

with engine.connect() as connection:
    # 先上传数据到临时表
    temp_table = f'#{tablename}_temp'
    data[requireddata].to_sql(temp_table, connection, index=False, if_exists='replace', schema='dbo')
    # 原子操作替换原表
    connection.exec_driver_sql(f"""
        BEGIN TRANSACTION
        DROP TABLE IF EXISTS dbo.{tablename};
        EXEC sp_rename '{temp_table}', '{tablename}';
        COMMIT TRANSACTION
    """)

3. 开启fast_executemany提升写入速度 #

关闭fast_executemany会导致逐条插入数据，拉长事务时间和锁持有周期。开启该参数可大幅提升批量写入效率，缩短锁的影响时间：

engine = sqlalchemy.create_engine(f'mssql+pyodbc:///?odbc_connect={connect_string}', fast_executemany=True)

4. 设置数据库锁超时（缓解方案） #

在数据库端配置锁超时，避免其他事务无限等待：

SET LOCK_TIMEOUT 5000; -- 设置超时时间为5秒，单位毫秒

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