sql-server – 使用SQLAlchemy从MS SQL到PostgreSQL的数据迁移

我想将数据从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和Mathias Fussenegger),它们采用了类似的方法来完成这项任务：

>连接到两个数据库
>反映源数据库的表
>迭代表格和每个表格

>在目标数据库中创建一个相等的表
>获取源中的行并将其插入目标数据库中

码

以下是使用上一个参考中的代码的简短示例.

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" (nt"idOperador" INTEGER NOT NULL,nt"idGrupo" INTEGER,nt"desigOperador" NVARCHAR(100) COLLATE "SQL_Latin1_General_CP1_CI_AS",nt"Rua" NVARCHAR(200) COLLATE "SQL_Latin1_General_CP1_CI_AS",nt"Localidade" NVARCHAR(200) COLLATE "SQL_Latin1_General_CP1_CI_AS",nt"codPostal" NVARCHAR(10) COLLATE "SQL_Latin1_General_CP1_CI_AS",ntdataini DATETIME,ntdataact DATETIME,ntemail NVARCHAR(50) COLLATE "SQL_Latin1_General_CP1_CI_AS",nturl NVARCHAR(50) COLLATE "SQL_Latin1_General_CP1_CI_AS",ntPRIMARY KEY ("idOperador")n)nn']

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

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

实际上,这个错误是从Urbanek运行代码获得的,我可以在其中指定要复制的表.运行上面的示例,让我…

问题2

MS SQL安装是使用ArcSDE(空间数据库引擎)的地理数据库.因此,某些列属于非defaultGeometry类型.在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进行进一步的模式迁移

我尝试过的其他事情都失败了(现在不记得确切的原因,但如果有任何答案,我会再次通过它们)：

>水壶
> Geokettle
> ogr2ogr(仍在尝试这种方法)

数据库细节：

>小型数据库,±3 GB
>±40桌
>有表格包含空间和非空间数据
>两个数据库(SQL Server和PostgreSQL)位于运行Windows Server 2008的同一服务器中
>停机时间没有大问题(最多8小时就可以了)

解决方法

可能这也适用于源数据库和目标数据库的其他组合(除了MS SQL Server和PostgreSQL之外),尽管它们没有经过测试.

工作流程(TL; DR)

>自动检查源并推导出现有的表模型(这称为反射).
>导入先前定义的表模型,这些表模型将用于在目标中创建新表.
>迭代表模型(源和目标中存在的模型).
>对于每个表,从源中获取行的块并将它们插入到目标中.

要求

> 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定义).

现在,如果你有十分之一的表,那么定义这么多表可能会令人困惑,乏味或容易出错.幸运的是,有一个工具可以读取现有数据库的结构并生成相应的SQLAlchemy模型代码.例：

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

因为此处的目的只是迁移数据而不是模式,所以您可以从源数据库创建模型,只需将生成的代码调整/更正到目标数据库.

注意：它将生成混合类模型和表模型.阅读here关于此行为.

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

>具有自定义数据类型的列使用NullType定义.用适当的类型替换它们,GeoAlchemy2的Geometry.
定义几何体时,传递正确的几何体类型(线串,多线串,多边形等)和SRID.
> PostgreSQL字符类型具有可变长度功能,SQLAlchemy默认将String列映射到它们,因此我们可以用String替换所有Unicode和String(…).请注意,它不是必需的,也不建议(不要引用我),指定String中的字符数,只需省略它们.
>您必须仔细检查,但是,可能所有BIT列实际上都是布尔值.
>对于字符类型,大多数数字类型(例如,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'x01x02x00x00x00x02x00x00x00@xb1xbfxec/xf8xf4xc0x80nF%x99(xf9xc0@xe3xa5x9bx94xf6xf4xc0x806xab>xc5%xf9xc0'))

此处的解决方法是将几何列转换为标准文本(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文档：

The returned result is an instance of ResultProxy,which references a
DBAPI cursor and provides a largely compatible interface with that of
the DBAPI cursor. The DBAPI cursor will be closed by the ResultProxy
when all of its result rows (if any) are exhausted.

只有在插入数据之前才会检索数据.

4.插入数据

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

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

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

4.1匹配列

首先,将源列与目标列(同一个表)进行匹配存在一些问题 – 可能这与“几何”列相关.一种可能的解决方案是创建一个Python字典,它将源列中的值映射到目标列的键(名称).

这是逐行执行的 – 尽管它并不像人们猜的那么慢,因为实际的插入将同时由几行组成.因此,每行将有一个字典,而不是插入数据对象(这是一个元组列表;一个元组对应一行),您将插入一个字典列表.

这是一行的示例.获取的数据是具有一个元组的列表,值是构建的字典.

# data
[(1,6,None,204,1,True,False,1.0,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中,几何数据类型不指定存储它的几何类型(即线,多边形等).它只关心它是一个几何.此信息存储在另一个(系统)表中,称为

The returned result is an instance of ResultProxy,which references a
DBAPI cursor and provides a largely compatible interface with that of
the DBAPI cursor. The DBAPI cursor will be closed by the ResultProxy
when all of its result rows (if any) are exhausted.

0000″ rel=”nofollow noreferrer”>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中插入数据时必须满足该条件.问题是当使用.STAsText()方法将几何数据作为WKT提取时,会丢失SRID信息.

幸运的是,PostGIS支持包含SRID的Extended-WKT(E-WKT)格式.
这里的解决方案是在修复几何时包含SRID.在相同的例子中,20 22)’被转换为’SRID = 102165; MULTILINESTRING((10 12,20 22))’.

4.4获取并插入

一切都修复后,您就可以插入了.如前所述,只有现在才能从源中实际检索数据.您可以在数据块(用户定义的数量)中执行此操作,一次1000行.

而真：
rows = data.fetchmany(1000)
如果不是行：
打破
values = [{key：(val if key.lower()！=“shape”else fix(val,102165))
用于键,用于拉链(键,行)的val}用于行中的行]
postgres_engine.execute(target_table.insert(),values)

这里fix()是一个函数,它将修正几何,并将给定的SRID添加到几何列(在本例中,通过“shape”的列名称标识) – 如上所述 – 并且值是上述词典列表.

结果

结果是存在于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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!