如何通过SQLAlchemy跨服务器/引擎实现表数据Insert from Select迁移

问题背景 #

已通过SQLAlchemy完成源/目标数据库引擎创建、源表反射、目标表结构复制，但数据迁移阶段希望实现无客户端中转的INSERT...SELECT式复制，当前暂用DataFrame中转，寻求更高效的方案。

解决方案 #

方案1：利用SQL Server链接服务器（真正无中转，推荐） #

此方案要求目标SQL Server能直接访问源服务器，需先在目标端配置链接服务器，数据在数据库层面直接复制，不经过客户端。

步骤1：在目标SQL Server创建链接服务器

执行以下SQL语句（替换占位符为实际信息）：

-- 创建链接服务器
EXEC sp_addlinkedserver 
   @server='SOURCE_LINKED_NAME',  -- 自定义链接服务器名称
   @srvproduct='',
   @provider='SQLNCLI11', 
   @datasrc='SOURCE_SERVER_IP_OR_NAME';  -- 源服务器地址/名称

-- 配置SQL认证映射（如果源服务器使用SQL账号）
EXEC sp_addlinkedsrvlogin 
   @rmtsrvname='SOURCE_LINKED_NAME',
   @useself='FALSE',
   @rmtuser='SOURCE_DB_USER',
   @rmtpassword='SOURCE_DB_PWD';

步骤2：用SQLAlchemy构造跨库INSERT语句

from sqlalchemy import insert, select, text

# 定义链接服务器、源库/架构/表的全限定名
source_remote_fullname = f"{linked_server_name}.{source_database}.{source_schema}.{source_object_name}"

# 构造INSERT...SELECT语句，确保列对应
insert_stmt = insert(destination_object).from_select(
    [col.name for col in destination_object.columns],
    select(*[text(f"{source_remote_fullname}.{col.name}") for col in destination_object.columns])
)

# 在目标引擎执行批量插入
with destination_engine.begin() as conn:
    conn.execute(insert_stmt)

方案2：流式查询+批量插入（无DataFrame中转，适合无法配置链接服务器场景） #

如果无法配置链接服务器，可通过流式查询逐批获取源数据，直接批量插入目标库，避免将全量数据加载到内存（如DataFrame）。

from sqlalchemy import select

# 构造源表查询语句
source_query = select(source_object)

# 流式读取源数据，批量插入目标库
batch_size = 1000  # 可根据内存情况调整批次大小
with source_engine.connect() as source_conn:
    # 开启流式结果，避免一次性加载全量数据
    result = source_conn.execution_options(stream_results=True).execute(source_query)
    
    while True:
        batch_data = result.fetchmany(batch_size)
        if not batch_data:
            break
        # 批量插入到目标库
        with destination_engine.begin() as dest_conn:
            dest_conn.execute(destination_object.insert(), batch_data)

方案对比 #

• 方案1：数据完全在数据库服务器间传输，效率最高，无客户端内存压力，是最优选择，但需数据库权限配置链接服务器。
• 方案2：数据经过客户端但采用流式处理，内存占用极低，无需数据库额外配置，适合权限受限场景。

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