使用 SQLAlchemy 从 MS SQL 到 PostgreSQL 的数据迁移

Question

TL;DR

我想将数据从 MS SQL Server + ArcSDE 迁移到 PostgreSQL + PostGIS，最好使用 SQLAlchemy。

---

我正在使用 SQLAlchemy 1.0.11 将现有数据库从 MS SQL 2012 迁移到 PostgreSQL 9.2（计划升级到 9.5）。

我一直在阅读这方面的内容，并找到了几个不同的来源（Tyler Lesmann、Inada Naoki、Stefan Urbanek 和 Stefan Urbanek 和 Mathias Fussenegger），它们采用了类似的方法来完成这项任务：

1. 连接到两个数据库
2. 反映源数据库的表
3. 遍历表并为每个表
   1. 在目标数据库中创建对等表
   2. 在源中获取行并将它们插入到目标数据库中

代码

这是一个使用上次参考代码的简短示例。

from sqlalchemy import create_engine, MetaData

src = create_engine('mssql://user:pass@host/database?driver=ODBC+Driver+13+for+SQL+Server')
dst = create_engine('postgresql://user:pass@host/database')

meta = MetaData()
meta.reflect(bind=src)

tables = meta.tables

for tbl in tables:
    data = src.execute(tables[tbl].select()).fetchall()
    if data:
        dst.execute(tables[tbl].insert(), data)

我知道同时获取所有行是一个坏主意，可以使用迭代器或fetchmany 来完成，但这不是我现在的问题。

问题 1

所有四个示例都在我的数据库中失败。我得到的错误之一与NVARCHAR 类型的列有关：

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) type "nvarchar" does not exist
LINE 5:  "desigOperador" NVARCHAR(100) COLLATE "SQL_Latin1_General_C...
                         ^
 [SQL: '\nCREATE TABLE "Operators" (\n\t"idOperador" INTEGER NOT NULL, \n\t"idGrupo" INTEGER, \n\t"desigOperador" NVARCHAR(100) COLLATE "SQL_Latin1_General_CP1_CI_AS", \n\t"Rua" NVARCHAR(200) COLLATE "SQL_Latin1_General_CP1_CI_AS", \n\t"Localidade" NVARCHAR(200) COLLATE "SQL_Latin1_General_CP1_CI_AS", \n\t"codPostal" NVARCHAR(10) COLLATE "SQL_Latin1_General_CP1_CI_AS", \n\tdataini DATETIME, \n\tdataact DATETIME, \n\temail NVARCHAR(50) COLLATE "SQL_Latin1_General_CP1_CI_AS", \n\turl NVARCHAR(50) COLLATE "SQL_Latin1_General_CP1_CI_AS", \n\tPRIMARY KEY ("idOperador")\n)\n\n']

我从这个错误的理解是PostgreSQL没有NVARCHAR而是VARCHAR，应该是等价的。我认为 SQLAlchemy 会自动将它们都映射到其抽象层中的String，但在这种情况下它可能不会那样工作。

问题：我是否应该事先定义所有的类/表，例如在models.py 中，以避免这样的错误？如果是这样，它将如何与给定（或其他）工作流程集成？

事实上，这个错误是在运行 Urbanek 的代码时得到的，我可以在其中指定要复制的表。运行上面的示例，我会...

问题 2

MS SQL 安装是使用ArcSDE（空间数据库引擎）的地理数据库。出于这个原因，一些列是非默认几何类型。在 PostgreSQL 方面，我使用的是 PostGIS 2。

尝试复制具有这些类型的表时，我收到如下警告：

/usr/local/lib/python2.7/dist-packages/sqlalchemy/dialects/mssql/base.py:1791: SAWarning: Did not recognize type 'geometry' of column 'geom'
  (type, name))
/usr/local/lib/python2.7/dist-packages/sqlalchemy/dialects/mssql/base.py:1791: SAWarning: Did not recognize type 'geometry' of column 'shape'

这些后面跟着另一个错误（这个错误实际上是在执行上面提供的代码时抛出的）：

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) relation "SDE_spatial_references" does not exist
LINE 1: INSERT INTO "SDE_spatial_references" (srid, description, aut...
                    ^

我认为它未能创建警告中提到的列，但是在需要这些列时在稍后的步骤中引发了错误。

问题：问题是上一个问题的扩展：如何使用自定义（或在其他地方定义）类型进行迁移？

我知道GeoAlchemy2 可以与 PostGIS 一起使用。 GeoAlchemy 支持 MS SQL Server 2008，但在这种情况下，我想我是 stuck with SQLAlchemy 0.8.4（也许没有那么好的功能）。另外，我发现here 可以使用 GeoAlchemy 定义的类型进行反射。但是，我的问题仍然存在。

可能相关

• https://stackoverflow.com/questions/34475241/how-to-migrate-from-mysql-to-postgressql-using-pymysql
• SqlAlchemy: export table to new database
• https://stackoverflow.com/questions/34956523/sqlalchemy-custom-column-type-use-bindparam-as-multiple-function-parameters
• SQLAlchemy Reflection Using Metaclass with Column Override

编辑

当我看到引用 SDE_spatial_references 的错误时，我认为这可能与 ArcSDE 有关，因为同一台机器上也安装了 ArcGIS for Server。然后我了解到 MS SQL Server 也有一些Spatial Data Types，然后我确认是这种情况。我这个编辑错了：数据库确实在使用 ArcSDE。

编辑 2

这里有一些我忘记包含的更多细节。

迁移不必使用 SQLAlchemy 完成。我认为这是个好主意，因为：

• 我更喜欢使用 Python
• 解决方案必须使用 FOSS
• 理想情况下，它应该是一种易于重现的方式，并且可以启动和等待
• 迁移后，我想使用 Alembic 进行进一步的架构迁移

我尝试过但失败的其他事情（现在不记得确切的原因，但如果有任何答案提到它们，我会再看一遍）：

• 水壶
• 地热水壶
• ogr2ogr（仍在尝试这种方法）

数据库详情：

• 小型数据库，± 3 GB
• ± 40 表
• 有空间和非空间数据的表
• 两个数据库（SQL Server 和 PostgreSQL）位于运行 Windows Server 2008 的同一服务器中
• 停机时间没有大问题（最多 8 小时即可）

Answer 1

我建议这个流程包含两个重要的迁移步骤：

迁移架构

• 转储源 DB 架构，最好是跨数据工具（如 UML）的某种统一格式（此步骤和后续步骤将需要并且更容易使用像 Toad Data Modeler 或 IBM Rational Rose 这样的收费）。
• 在需要时使用 TDM 或 RR 将表定义从源类型更改为目标类型。例如。在 postgres 中摆脱 varchar(n) 并坚持使用 text，除非您特别需要应用程序在字符串长于 n 的数据层上崩溃。如果无法在数据建模工具中进行转换，请忽略（暂时）几何等复杂类型。
• 为目标数据库生成一个DDL-file（再次提到的工具在这里很方便）。
• 创建（并添加到表中）复杂类型，因为它们应该由目标 RDBMS 处理。尝试插入几个条目以确保数据类型一致。 将这些类型添加到您的 DDL 文件中。
• 您可能还想在此处禁用外键约束等检查。

迁移数据

1. 将简单表（即带有标量字段）转储到 CSV。
2. Import simple tables data。
3. 编写一段简单的代码来从源中选择复杂数据并将其插入到目标中（这比听起来容易，只需选择 -> 映射属性 -> 插入）。不要在一个代码例程中编写所有复杂类型的迁移，保持简单，分而治之。
4. 如果您在迁移架构时没有禁用检查，则可能需要对不同的表重复第 2 步和第 3 步（这就是为什么禁用检查:)）。
5. 启用检查。

通过这种方式，您可以将迁移过程拆分为简单的原子步骤，并且数据迁移第 3 步的失败不会导致您移回架构迁移等。您只需截断几个表，然后重新运行失败时数据导入。

Answer 2

这是我使用 SQLAlchemy 的解决方案。这是一篇类似长博客的帖子，我希望它在这里可以接受，并且对某人有用。

这可能也适用于 源 和 目标 数据库的其他组合（分别除了 MS SQL Server 和 PostgreSQL），尽管它们没有经过测试。

工作流程（某种 TL；DR）

1. 自动检查源并推断现有表模型（这称为反射）。
2. 导入之前定义的表模型，这些模型将用于在目标中创建新表。
3. 迭代表模型（源和目标中都存在的模型）。
4. 对于每个表，从源中获取行块并将它们插入到目标中。

---

要求

• SQLAlchemy
• GeoAlchemy2
• sqlacodegen

---

详细步骤

1。连接数据库

SQLAlchemy 向处理应用程序和实际数据库之间连接的对象调用引擎。因此，要连接到数据库，必须使用相应的连接字符串创建引擎。数据库 URL 的典型形式是：

dialect+driver://username:password@host:port/database

您可以在SQLAlchemy documentation 中看到一些连接 URL 的示例。

一旦创建，引擎将不会建立连接，除非通过.connect() 方法或调用依赖此方法的操作（例如，.execute()）明确告知它这样做。

con = ms_sql.connect()

2。定义和创建表

2.1 源数据库

源端的表已经定义好了，所以我们可以使用表反射：

from sqlalchemy import MetaData

metadata = MetaData(source_engine)
metadata.reflect(bind=source_engine)

如果您尝试这样做，您可能会看到一些警告。例如，

SAWarning: Did not recognize type 'geometry' of column 'Shape'

这是因为 SQLAlchemy 不能自动识别自定义类型。在我的具体情况下，这是因为 ArcSDE 类型。但是，当您只需要读取数据时，这不是问题。忽略这些警告即可。

表反射后，您可以通过该元数据对象访问现有表。

# see all the tables names
print list(metadata.tables)
# handle the table named 'Troco'
src_table = metadata.tables['Troco']
# see that table columns
print src_table.c

2.2 目标数据库

对于目标，因为我们正在启动一个新的数据库，所以不可能使用表反射。但是，创建表模型through SQLAlchemy并不复杂；事实上，它可能比编写纯 SQL 还要简单。

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class SomeClass(Base):
    __tablename__ = 'some_table'
    id = Column(Integer, primary_key=True)
    name =  Column(String(50))
    Shape = Column(Geometry('MULTIPOLYGON', srid=102165))

在此示例中，有一列包含空间数据（此处由 GeoAlchemy2 定义）。

现在，如果您有十分之一的表，那么定义这么多表可能会令人费解、乏味或容易出错。幸运的是，有sqlacodegen，这是一个读取现有数据库结构并生成相应 SQLAlchemy 模型代码的工具。示例：

pip install sqlacodegen
sqlacodegen mssql:///some_local_db --outfile models.py

因为这里的目的只是迁移数据，而不是架构，所以您可以从源数据库创建模型，并将生成的代码调整/更正到目标数据库。

注意：它将生成混合的class 模型和Table 模型。阅读here 了解此行为。

同样，您会看到关于无法识别的自定义数据类型的类似警告。这就是我们现在必须编辑 models.py 文件并调整模型的原因之一。以下是一些需要调整的提示：

• 具有自定义数据类型的列使用NullType 定义。将它们替换为正确的类型，例如 GeoAlchemy2 的 Geometry。 定义 Geometry 时，传递正确的几何类型（线串、多线串、多边形等）和 SRID。
• PostgreSQL 字符类型支持可变长度，SQLAlchemy 默认将String 列映射到它们，因此我们可以将所有Unicode 和String(...) 替换为String。请注意，在String 中指定字符数不是必需的，也不可取（不要引用我的话），只需省略它们即可。
• 您必须仔细检查，但很可能所有BIT 列实际上都是Boolean。
• 大多数数字类型（例如，Float(...)、Numeric(...)），同样对于字符类型，可以简化为Numeric。小心例外和/或某些特定情况。
• 我注意到定义为索引的列存在一些问题 (index=True)。就我而言，由于架构将被迁移，因此现在不需要这些，可以安全地删除。
• 确保两个数据库（反射表和定义模型）中的表名和列名相同，这是后续步骤的要求。

现在我们可以将模型和数据库连接在一起，并在目标端创建所有表。

Base.metadata.bind = postgres
Base.metadata.create_all()

请注意，默认情况下，.create_all() 不会触及现有表。如果您想重新创建或向现有表中插入数据，则需要事先DROP。

Base.metadata.drop_all()

3。获取数据

现在您可以从一侧复制数据，然后将其粘贴到另一侧。基本上，您只需要为每个表发出一个SELECT 查询。这在 SQLAlchemy ORM 提供的抽象层上是可能且容易做到的。

data = ms_sql.execute(metadata.tables['TableName'].select()).fetchall()

但是，这还不够，您还需要更多的控制权。其原因与 ArcSDE 有关。因为它使用专有格式，所以您可以检索数据，但无法正确解析它。你会得到这样的东西：

(1, Decimal('0'), u' ', bytearray(b'\x01\x02\x00\x00\x00\x02\x00\x00\x00@\xb1\xbf\xec/\xf8\xf4\xc0\x80\nF%\x99(\xf9\xc0@\xe3\xa5\x9b\x94\xf6\xf4\xc0\x806\xab>\xc5%\xf9\xc0'))

这里的解决方法是将几何列转换为众所周知的文本 (WKT) 格式。这种转换必须在数据库端进行。 ArcSDE 在那里，所以它知道如何转换它。因此，例如，在 TableName 中有一个名为 shape 的包含空间数据的列。所需的 SQL 语句应如下所示：

SELECT [TableName].[shape].STAsText() FROM [TableName]

这使用.STAsText()，SQL Server 的几何数据类型方法。

如果您不使用 ArcSDE，则不需要执行以下步骤：

• 遍历表（仅在源和目标中定义的表），
• 对于每个表，查找几何列（事先列出）
• 像上面那样构建一条 SQL 语句

一旦构建了语句，SQLAlchemy 就可以执行它。

result = ms_sql.execute(statement)

事实上，这实际上并没有获取数据（与 ORM 示例相比——请注意缺少的 .fetchall() 调用）。为了解释，这里引用了 SQLAlchemy 文档：

返回的结果是ResultProxy的一个实例，它引用了一个 DBAPI 游标，并提供与 DBAPI 游标。 DBAPI 游标将被ResultProxy 关闭 当其所有结果行（如果有）都用尽时。

数据只会在插入之前被检索。

4。插入数据

连接已建立，表已创建，数据已准备好，现在让我们插入它。与获取数据类似，SQLAlchemy 还允许通过其 ORM 将INSERT 数据放入给定的表中：

postgres_engine.execute(Base.metadata.tables['TableName'].insert(), data)

同样，这很容易，但由于非标准格式和错误数据，可能需要进一步操作。

4.1 匹配列

首先，将源列与目标列（同一个表的）匹配存在一些问题——这可能与Geometry 列有关。一种可能的解决方案是创建一个 Python 字典，它将源列中的值映射到目标列的键（名称）。

这是逐行执行的——尽管它并不像人们想象的那么慢，因为实际插入将同时插入几行。因此，每行将有一个字典，并且您将插入一个字典列表，而不是插入数据对象（这是一个元组列表；一个元组对应一行）。

这是一个单行的示例。获取的数据是一个包含一个元组的列表，值是构建的字典。

# data
[(1, 6, None, None, 204, 1, True, False, 204, 1.0, 1.0, 1.0, False, None]
# values
[{'DateDeleted': None, 'sentidocirculacao': False, 'TempoPercursoMed': 1.0,
  'ExtensaoTroco': 204, 'OBJECTID': 229119, 'NumViasSentido': 1,
  'Deleted': False, 'TempoPercursoMin': 1.0, 'IdCentroOp': 6,
  'IDParagemInicio': None, 'IDParagemFim': None, 'TipoPavimento': True,
  'TempoPercursoMax': 1.0, 'IDTroco': 1, 'CorredorBusext': 204}]

请注意，Python 字典没有排序，这就是为什么两个列表中的数字不在同一位置的原因。为简化起见，此示例中删除了几何列。

4.2 固定几何形状

如果没有发生此问题，则可能不需要以前的解决方法：有时几何图形的存储/检索类型错误。

在 MSSQL/ArcSDE 中，几何数据类型不指定存储的几何类型（即线、多边形等）。它只关心它是一个几何。此信息存储在另一个（系统）表中，称为SDE_geometry_columns（参见该页底部）。但是，Postgres（实际上是 PostGIS）在定义几何列时需要几何类型。

这会导致空间数据以错误的几何类型存储。错误的意思是它与应有的不同。例如，查看 SDE_geometry_columns 表（摘录）：

f_table_name        geometry_type
TableName               9

geometry_type = 9 对应于ST_MULTILINESTRING。但是，TableName 表中有一些行存储（或接收）为ST_LINESTRING。这种不匹配会在 Postgres 端引发错误。

作为一种解决方法，您可以在创建上述词典时编辑 WKT。例如，'LINESTRING (10 12, 20 22)' 转换为 MULTILINESTRING ((10 12, 20 22))'。

4.3 缺少 SRID

最后，如果您愿意保留 SRID，您还需要在创建几何列时定义它们。

如果表模型中定义了 SRID，则在 Postgres 中插入数据时必须满足该 SRID。问题是当使用.STAsText() 方法将几何数据作为WKT 获取时，您会丢失SRID 信息。

幸运的是，PostGIS 支持包含 SRID 的 Extended-WKT (E-WKT) 格式。 此处的解决方案是在修复几何图形时包含 SRID。同样的例子，'LINESTRING (10 12, 20 22)' 被转换为'SRID=102165;MULTILINESTRING ((10 12, 20 22))'。

4.4 获取和插入

一切就绪后，您就可以插入了。如前所述，只有现在才能真正从源中检索数据。您可以在数据块（用户定义的数量）中执行此操作，例如，一次1000 行。

当真时： 行 = data.fetchmany(1000) 如果不是行： 休息 values = [{key: (val if key.lower() != "shape" else fix(val, 102165)) for key, val in zip(keys, row)} for row in rows] postgres_engine.execute(target_table.insert(), values)

这里的fix() 是一个函数，它将校正几何并将给定的 SRID 附加到几何列（在本例中，由“shape”的列名标识）——如上所述——和values 是前面提到的字典列表。

结果

结果是将 MS SQL Server + ArcSDE 数据库上的架构和数据复制到 PostgreSQL + PostGIS 数据库中。

以下是我的用例中的一些统计数据，用于性能分析。两个数据库在同一台机器上；代码是从不同的机器上执行的，但在同一个本地网络中。

Tables   |   Geometry Column   |   Rows   |   Fixed Geometries   |   Insert Time
---------------------------------------------------------------------------------
Table 1      MULTILINESTRING      1114797             702              17min12s
Table 2            None            460874             ---               4min55s
Table 3      MULTILINESTRING       389485          389485               4min20s
Table 4        MULTIPOLYGON          4050            3993                   34s
Total                             3777964          871243              48min27s

Answer 3

我在尝试从 Oracle 9i to MySQL 迁移时遇到了同样的问题。

我构建了etlalchemy 来解决这个问题，目前它已经在MySQL、PostgreSQL、SQL Server、Oracle 和SQLite 之间进行了迁移测试。它利用了上述 RDBMS 的 SQLAlchemy 和 BULK CSV Import 功能（而且速度非常快！）。

安装（非 El-capitan）：pip install etlalchemy

安装（El-capitan）：pip install --ignore-installed etlalchemy

运行：

from etlalchemy import ETLAlchemySource, ETLAlchemyTarget
# Migrate from SQL Server onto PostgreSQL
src = ETLAlchemySource("mssql+pyodbc://user:passwd@DSN_NAME")
tgt = ETLAlchemyTarget("postgresql://user:passwd@hostname/dbname",
                          drop_database=True)
tgt.addSource(src)
tgt.migrate()