使用SQLAlchemy迁移SQL Server表元数据到PostgreSQL的问题咨询

问题原因 #

你直接从MSSQL反射得到的表对象绑定的是MSSQL方言专属数据类型（比如tinyint），这类类型没有做跨数据库适配，PostgreSQL方言无法解析，因此建表报错。

可行解决方法 #

• 方案1：反射阶段自动做类型映射
  适合多表批量迁移的场景，一次配置全表生效：

from sqlalchemy import create_engine, MetaData, Table, SmallInteger, String, Text, TIMESTAMP
from sqlalchemy.dialects.mssql import TINYINT, NVARCHAR, SMALLDATETIME, BIT
from sqlalchemy import event
import pandas as pd

engine_pg = create_engine('postgresql://XXXX:YYYY$@10.10.1.4:5432/pgschema')
engine_ms = create_engine('mssql+pyodbc://XX:YY@10.10.1.5/msqlschema?driver=SQL+Server')
ms_metadata = MetaData(bind=engine_ms)

# 注册反射列的监听事件，自动替换不兼容类型
@event.listens_for(ms_metadata, 'column_reflect')
def convert_types(inspector, table, column_info):
    # MSSQL TINYINT 映射为PG支持的SmallInteger
    if isinstance(column_info['type'], TINYINT):
        column_info['type'] = SmallInteger()
    # MSSQL 超长字符串映射为PG的TEXT
    elif isinstance(column_info['type'], NVARCHAR) and column_info['type'].length is None:
        column_info['type'] = Text()
    # MSSQL SMALLDATETIME映射为PG的TIMESTAMP
    elif isinstance(column_info['type'], SMALLDATETIME):
        column_info['type'] = TIMESTAMP()
    # MSSQL BIT映射为PG的BOOLEAN
    elif isinstance(column_info['type'], BIT):
        from sqlalchemy import Boolean
        column_info['type'] = Boolean()

# 反射表时就会自动做类型转换
Node = Table('Node', ms_metadata, autoload_with=engine_ms)
# 直接在PG创建即可
Node.create(bind=engine_pg)

• 方案2：反射后手动修改列类型
  适合少量表临时调整的场景：

from sqlalchemy.dialects.mssql import TINYINT
from sqlalchemy import SmallInteger

Node = Table('Node', ms_metadata, autoload_with=engine_ms)
# 遍历列替换不兼容类型
for col in Node.columns:
    if isinstance(col.type, TINYINT):
        col.type = SmallInteger()

Node.create(bind=engine_pg)

• 方案3：借助Pandas中转建表
  适合小表快速迁移的场景：

# 只读取表结构不读数据
df = pd.read_sql("SELECT TOP 0 * FROM Node", engine_ms)
# 自动适配PG类型建表
df.to_sql('Node', engine_pg, if_exists='replace', index=False)

跨数据库迁移最佳实践 #

• 提前梳理全量类型映射规则：先把两个数据库的类型对应关系明确，比如MSSQL UNIQUEIDENTIFIER → PG UUID、MSSQL NTEXT → PG TEXT等，统一在反射监听事件里配置，避免逐个表修改。
• 结构校验优先：建表完成后先对比两边的列数量、列名、非空约束、主键、索引是否一致，再开始迁移数据。
• 大表迁移优化：全量导入前先关闭PG端的索引、外键约束，数据导入完成后再重建，可提升导入速度3-10倍；分批拉取数据避免内存溢出。
• 数据一致性校验：迁移完成后对比两边表的行数、数值列的求和/极值、字符串列的哈希聚合值，确保数据没有丢失或转换错误。
• 增量迁移方案：如果是线上业务不停机迁移，可先全量迁移历史数据，再通过增量日志同步增量数据，最终切流验证。

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