国产化数据库迁移工具不会用？教你手搓一个万能数据迁移工具。

为什么要手搓一个自己的数据库迁移工具

为什么要进行数据库迁移？主要有这么几种情况：

（1）、开发测试阶段使用的数据库类型于生产环境的数据库类型不同，如开发测试用MySQL数据库，生产环境用Oracle\SQLServer等企业级数据库，需要将开发测试环境的一些基础数据迁移到生产环境；为什么开发测试用的数据库类型与生产环境不同，自然是为了降低开发成本，生产环境的数据库是客户购买的，软件开发方未必能有条件购买使用同类型的数据库产品。

（2）、开发测试环境使用的数据库版本与生产环境的数据库不同，例如开发测试环境使用的是数据库的“开发版”、社区版，或者较低的版本，生产环境用的是企业版、最新的版本；这样做自然也是为了降低开发成本，比如开发环境用金仓V8版本，而线上用的是最新的金仓V9版本。

（3）、国产化要求，需要将之前运行在MySQL、Oracle、SQLServer等数据库上的数据迁移到金仓、达梦等国产数据库上；国产化要求是现在很多政府类项目的硬性要求，软件公司之前成熟的产品用的是非国产化数据库，现在都有数据迁移需求。

（4）、提高数据库性能和数据备份的要求；线上一些系统运行时间长，产生的很多历史数据需要迁移到其它数据库上以便给线上的数据库“瘦身”，起到一个数据备份和提高线上系统查询效率的作用；比如一些日志数据、报警数据，可以把归档后的数据迁移到新的数据库上。

在国产数据库领域，金仓数据库算是比较常用的一种数据库了，号称百分百兼容Oracle的，也基本百分百兼容MySQL等数据库，能“无缝”迁移。金仓等国产数据库在兼容性这方面的确很努力，如果这事不努力国人为什么要用你的数据库？在实际使用过程中的确感觉不到兼容性问题，但在做数据迁移的时候才发现在兼容性方面还是有一些问题，而金仓提供的数据迁移工具对于非资深用户而言还是有较高的门槛，每次遇到问题都不得不求助于金仓的技术支持人员，每次寻求他们支持都要沟通很久，于是我决定自己手搓一个数据库迁移工具，而且还能反向迁移，即从金仓迁移到MySQL等各种数据库，而不是金仓自己的工具只能从Oracle，SQL Server,MySQL,PostgreSQL这几种数据库迁移到金仓。

关于数据迁移，以前工作中经常进行，因此积累了一定的经验，而且还把这些经验写到《SOD框架“企业级”应用数据架构实战》这本书里面了。下面介绍一下怎么使用SOD框架手搓一个数据迁移工具，开始之前必须先了解数据迁移有哪些问题，才能明白手搓一个自己的迁移工具的好处。

数据迁移的常见问题

数据迁移并不是一个简单的工作，虽然很多数据库都提供了将别的数据库迁移过来的数据迁移工具，但使用的时候还是会遇到很多问题：

第一个问题是数据量很大，我们往往希望把一个有几百万行、几千万行的表数据从一个小型数据库迁移到一个大中型的数据库中，比如把一个单机MySQL数据库中的数据迁移到金仓数据库集群中；或者将上百万的数据迁移到历史数据库中存档给原来的库瘦身。

第二个问题就是不同类型数据库之间进行迁移，比如从MySQL迁移到金仓、从金仓迁移到Oracle，由于源数据库和目标数据库类型不同，它们支持的字段类型、表类型甚至SQL语法都有差异。有些数据库厂商从商业上考虑可能只愿意提供将别的数据库迁移到自家数据库来的功能而不支持反向迁移。

第三个问题是同一种数据库不同版本之间进行数据迁移。按理说数据库不同版本之间应该保持兼容，至少是高版本兼容低版本的，但国产数据库很牛，它不同版本之间的数据类型是可能不兼容的，导致数据无法直接迁移，这个问题让我明白了金仓数据库迁移工具为何使用起来那么复杂。

第四个问题是数据迁移过程可能还伴随数据筛选、数据清洗和转换，这个属于ETL的范畴了，有一些成熟的ETL工具可以使用，但这些工具使用复杂并且不一定免费。

接下来我们自己手搓迁移工具时看怎么解决上面这些问题。

迁移方案设计

迁移的数据量很大，所以不能读取太多的数据到内存，最好从源数据读取一部分就写入目标数据库一部分数据；
不同种类数据库之间进行数据迁移，由于数据库之间SQL语法、字段类型有差异，所以最好不要直接采用编写SQL语句的方式来实现，用ORM框架可以完美解决这个问题；
数据库不同版本之间的数据迁移，注意采用兼容的数据类型即可，如果不能兼容也有办法；
第四个问题好办了，由于是自己手搓的工具，迁移前后可以自定义自己的处理逻辑，进行数据筛选、清洗和转换工作都不在话下了。

根据这个迁移方案，采用SOD框架来实现是很合适的，它的一些特性解决这些问题具有很大优势，下面我们逐个介绍怎么实现。

准备工作

首先我们需要明确源数据库和目标数据库的类型、版本，数据库连接信息，要迁移的表和视图数据。比如本文的例子以从金仓数据库迁移到MySQL数据库为例，使用VS先创建一个控制台项目，目标框架选择.NET8，然后项目中添加两个Nuget包，在目文件中添加下面的包引用代码：

 <PackageReference Include="PWMIS.SOD.Kingbase.Provider.Net6V9" Version="6.0.1" /><PackageReference Include="PWMIS.SOD.MySQL.Provider" Version="6.0.3" />

或者使用Nuget包管理工具，查找 PWMIS.SOD 关键字，然后安装SOD框架的金仓数据库访问提供程序和MySQL数据库访问提供程序：

Install-Package PWMIS.SOD.Kingbase.Provider
Install-Package PWMIS.SOD.MySQL.Provider

然后在解决方案资源管理器选择项目名称，右键菜单“添加-新建项-常规”，然后选择“应用程序配置文件”，添加一个 app.config文件，内容如下：

<?xml version="1.0" encoding="utf-8"?>
<configuration><appSettings><!--PDF.NET.SOD SQL 日志记录配置（for 4.0）开始记录执行的SQL语句，关闭此功能请将SaveCommandLog 设置为False，或者设置DataLogFile 为空；如果DataLogFile 的路径中包括~符号，表示SQL日志路径为当前Web应用程序的根目录；如果DataLogFile 不为空且为有效的路径，当系统执行SQL出现了错误，即使SaveCommandLog 设置为False，会且仅仅记录出错的这些SQL语句；如果DataLogFile 不为空且为有效的路径，且SaveCommandLog 设置为True，则会记录所有的SQL查询。在正式生产环境中，如果不需要调试系统，请将SaveCommandLog 设置为False 。--><add key="SaveCommandLog" value="True" /><add key="DataLogFile" value="Log\SqlLog.txt" /><!--LogExecutedTime 需要记录的时间，如果该值等于0会记录所有查询，否则只记录大于该时间的查询。单位毫秒。--><add key="LogExecutedTime" value="500" /><!--PDF.NET SQL 日志记录配置 结束--></appSettings><connectionStrings><add name="SourceDb"connectionString="Server=127.0.0.1;User Id=system;Password=system;Port=54321;Database=mydb;"providerName="PWMIS.DataProvider.Data.Kingbase,PWMIS.KingbaseClient.Net6V9" /><add name ="TargetDb"connectionString="server=127.0.0.1;User Id=root;password=123456;DataBase=mydb;"providerName="PWMIS.DataProvider.Data.MySQL,PWMIS.MySqlClient" /></connectionStrings></configuration>

View Code

有关如何配置连接字符串的详细内容，请移步框架的Nuget下载页面：NuGet Gallery | PWMIS.SOD 6.0.3

注意金仓数据库访问程序的选择不同版本有点差异，可以移步SOD的金仓数据库Nuget下载页面详细了解：NuGet Gallery | PWMIS.SOD.Kingbase.Provider 6.0.7

到此使用SOD框架开发数据迁移工具的准备工作已经完成，下面正式开始编写实现代码。

创建目标数据库

数据迁移通常都是目标数据库已经存在的情况下进行的，但这里为什么要强调创建目标数据库呢？这是因为很可能既有的目标数据库的数据表和表字段与源数据库是不兼容的，比如目标数据库的字符编码是UTF8，而源数据库是GB2312，目标表的字段类型是int而源表字段的类型是long,或者表字段都是varchar类型但是源表和目标表该字段的长度却不相同，当然更夸张的是连字段名都可能不相同（字段业务含义是一样的），这些千奇百怪的问题只有你想不到没有你遇不到的。所以最佳办法是由迁移工具自动创建一个目标数据库。

SOD框架的Code First方案可以由实体类创建表，它在第一次连接数据库的时候检查表是否存在，如果不存在才创建表，如果表已经存在则跳过以避免意外更改表结构。实现此过程很简单，只需要继承DbContext即可，比如对于本文的目标数据库，创建一个TargetDbContext类：

public class TargetDbContext : DbContext
{public TargetDbContext () : base("TargetDb"){}protected override bool CheckAllTableExists(){CheckTableExists<UserInfo>();//创建其它表。。。return true;}
}

在上面的代码中，DbContext类的构造函数参数值“TargetDb” 就是app.config中配置的连接名称，重载方法CheckAllTableExists 中 CheckTableExists泛型方法的类型参数UserInfo是一个SOD实体类，它可以根据实体类指定的表名称来创建目标表。这样，当TargetDbContext类型的对象被实例化的时候就会自动创建好迁移数据的目标表了。

迁移标识字段

数据库的标识字段是用来唯一标识一行数据的，主键就起到这种作用，我们也使用带自增功能的字段做主键，但自增字段不一定都是主键，本文说的标识字段是数据库的自增标识列，如SQLServer的IDENTITY 列，MySQL 用 AUTO_INCREMENT，Oracle 用 SEQUENCE+触发器或 IDENTITY 列，PostgreSQL和金仓数据库也是用 SEQUENCE 并设置 DEFAULT nextval。

SOD框架的Code First功能可以为各种数据库自动创标识列，只需要实体类设置 Identity="标识字段名"即可，示例代码如下：

public class UserInfo : EntityBase
{public UserInfo(){TableName = "UserInfo";IdentityName = "ID";    //标识字段PrimaryKeys.Add("ID"); //主键
  }public int ID{get{return getProperty<int>("ID");}set{setProperty("ID",value );}}public string Name{get{return getProperty<string>("Name");}set{setProperty("Name",value ,50);}}
}

默认情况下自增字段（IDENTITY / AUTO_INCREMENT / SERIAL）在插入数据的时候不能直接插入值，但在数据迁移的时候，需要将自增字段的值也迁移过去，除非自增字段没有被别的表在逻辑上引用。如果确实需要给自增列塞一个指定值，必须显式关闭/绕过自增字段的这个机制，操作完恢复自增字段的默认行为，否则后续普通方式插入数据会出错。比如对于SQLServer数据库：

-- 1 允许手动插入自增字段值
SET IDENTITY_INSERT dbo.UserInfo ON;
-- 2 手动写值（列清单必须显式写出）
INSERT INTO dbo.UserInfo (ID Name) VALUES (100, 'zhangsan');
-- 3 恢复自增字段默认行为
SET IDENTITY_INSERT dbo.UserInfo OFF;

对于PostgreSQL和金仓数据库的默认模式（PG模式）下，自增字段可以直接插入值，只要插入的值与现有自增字段值不重复即可。SOD框架根据实体类是否设置IdentityName属性来决定插入数据的时候是否插入自增列的值，所以在使用SOD框架迁移数据的时候除了要注意目标数据库对于自增字段的问题，还需要设置IdentityName属性为空值，我们定义一个 IImportable 接口来表示该实体类可以插入自增字段值：

 public interface IImportable{void IgnoreIdentity();}

修改前面的实体类，将IdentityName设置为空：

public class UserInfo : EntityBase,IImportable
{public public_AlarmsInfo(){TableName = "UserInfo";IdentityName = "ID";     //标识字段PrimaryKeys.Add("ID"); //主键
  }public void IgnoreIdentity(){IdentityName = "";}public int ID{get{return getProperty<int>("ID");}set{setProperty("ID",value );}}public string Name{get{return getProperty<string>("Name");}set{setProperty("Name",value ,50);}}
}

采用这种方式在运行时修改IdentityName 属性值，既可以享受到Code First自动创建目标表的便利，又可以实现插入自增列数据的功能；注意迁移完自增列数据后，需要重置自增列的标识数据到最大的自增列值，这样后续插入数据才不会出问题。对于金仓数据库迁移完成当前表的数据后，可以用下面的方式重置自增列的标识数据：

//entity 是当前正在迁移的表对于的SOD实体类对象if (targetDb.CurrentDBMSType == PWMIS.Common.DBMSType.Kingbase){//更新序列值ALTER SEQUENCE equipment_id_seq RESTART WITH 100;string tableName = entity.GetTableName();string sql = $"ALTER SEQUENCE {tableName.ToLower()}_id_seq RESTART WITH {max_id + 1}";targetDb.ExecuteNonQuery(sql);}

另外一种可选方式是在数据库上直接将原来的标识字段修改为普通字段，然后在实体类构造函数里面注释掉 IdentityName 这行字段即可，但用这种方式来进行Code First模式开发无法自动创建自增列，但可以等数据迁移完成后再手动设置自增标识。

大数据量查询

一次性在内存中加载10万条数据很可能导致进程无法正常运行，而且这种大数据也会导致.NET内存难以有效回收，而大表数据迁移又是很常见的事情，所以最佳方案是数据逐条读取，读一部分写一部分，避免将大量数据读取到内存后再写入，这样可以加快迁移速度。SOD框架的实体类查询支持这种“迭代器查询”，通过调用EntityQuery<T>.QueryEnumerable方法：

static void DataMigration<T>(AdoHelper sourceDb,AdoHelper targetDb,Action<T> action, string identityName="ID") where T : EntityBase, new()
{
//其它代码略var oql = OQL.From<T>().END;var readQuery = EntityQuery<T>.QueryEnumerable(oql, sourceDb);var insetQuery = new EntityQuery<T>(targetDb);foreach (var item in readQuery){action(item);//写入数据到目标数据库，代码暂略
 }//其它代码略
}

QueryEnumerable 方法通过DataReader对象循环读取数，每次只返回一个读取的实体类对象，从而避免了一次读取大量数据的问题。

数据复制

广义的数据复制是将读取的数据写入到目标数据库，但这里说的数据复制是将上面读取的数据复制到一个新的对象里面。虽然理论上可以将从源数据库读取的实体类直接写入到目标数据库，但数据迁移的环境可能比较复杂，比如源数据库和目标数据库是不同类型的数据库，或者虽然类型一样但是版本不一样，或者字段名称不一样甚至字段类型都不完全一样；另外一个原因是SOD实体类的设计与市面上绝大部分ORM都不同，SOD实体类采用值数组的方式存储从数据库读取的原始值，这些值可能携带了数据驱动程序特定的类型信息，而这种类型可能与目标数据库的类型是不兼容的，比如日期类型，MySQl驱动程序有自己的日期子类型，金仓数据库驱动程序也有自己的日期子类型，甚至不同版本的金仓数据库日期子类型还有微小的差异，所以数据迁移的时候最好消除源数据库读取字段的特定的类型信息，直接使用.NET的数据类型，然后让数据库驱动程序根据.NET数据类型转换到目标数据库支持的数据类型。驱动程序数据类型转换的问题比较复杂这里不细究。

针对不同的数据复制场景，SOD有不同的支持方案，最通常的方案是直接调用实体类的MapForm方法做数据映射拷贝：

 var insetQuery = new EntityQuery<T>(targetDb);foreach (var item in readQuery){T targetEntity = new T();targetEntity.MapFrom(item,false);targetEntity.ResetChanges(true);//其它代码略}

上面代码中MapFrom方法表示从任意一个实例对象中拷贝与当前实体类同名属性的值到当前实体类中，ResetChanges方法强行设置所有属性的修改状态是否修改，SOD框架会根据实体类属性是否修改（是否进行过赋值操作）来决定是否将该属性的值更新或者插入到数据库中。除了调用上面的两个方法，直接使用SOD的扩展方法CopyTo方法也可以实现类似的效果：

foreach (var item in readQuery)
{T targetEntity = new T();item.CopyTo(targetEntity);
}

但是使用上面的方式都没法复制源数据库表字段的NULL值，这个功能对SOD框架来说很简单：

foreach (var item in readQuery)
{T targetEntity = new T();item.CopyTo(targetEntity);for (int i = 0; i < item.PropertyValues.Length; i++){if (item[i] == DBNull.Value){targetEntity.PropertyValues[i]= DBNull.Value;}}
}

上面的代码中item是源数据库的实体类对象，targetEntity是目标数据库的实体类对象，SOD的实体类具有索引器功能，可以通过索引访问属性值，也可以通过索引直接修改实体类的属性值，这样就可以为属性设置NULL值。实体类的这种访问方式是绝大部分ORM框架都不支持的功能，这个功能为SOD框架处理数据带来了极大的便利性。

批量插入

目标数据库的写入是数据迁移过程中最慢的操作，批量插入能大大提高插入操作的性能，很多数据库都有一次性插入多条数据的功能，其中大多支持下面这种方式：

INSERT INTO 表名 (列1, 列2, …)
VALUES(值1_1, 值1_2, …),(值2_1, 值2_2, …),…(值n_1, 值n_2, …);

SOD框架对于MySQL和金仓数据库采用了这种方式进行批量插入，对于SQLServer采用了SqlBulkCopy方案。只要支持批量插入，都可以调用QuickInsert方法：

InsertListData<T>(List<T> targetList, EntityQuery<T> insetQuery) where T : EntityBase, new()
{int importedCount=insetQuery.QuickInsert(targetList);return importedCount;
}

如果数据库不支持批量插入，也可以使用EntityQuery对象的Insert重载方法插入一个实体列表对象，内部使用事务等方式优化插入性能。

进度信息

数据迁移可能耗时比较长，迁移过程中实时显示迁移进度是必要的，我实现了一个ConsoleProcessDisplayer类，效果类似Linux系统中下载文件的命令行进度显示方式，下面直接给出主要代码：

/// <summary>
/// 显示处理进度
/// </summary>
/// <param name="readIndex">读取的数据位置</param>
/// <param name="writeIndex">写入的数据位置</param>
public void DisplayProcessing(int readIndex,int writeIndex=0)
{if(writeIndex==0) writeIndex = readIndex;if (readIndex >= recordCount){Console.SetCursorPosition(this.left, this.top);var str = new string('=', 100);Console.Write("{0}[R:{1}/W:{2}/C:{3}]({4}%)", str, readIndex, writeIndex, recordCount, 100);return;}int currMSec = DateTime.Now.Millisecond / 100; //0.1秒的显示间隔if(currMSec!=lastMSec){lastMSec = currMSec;//显示控制Console.SetCursorPosition(this.left, this.top);char[] w_arr = new char[screenWidth];for (int j = 0; j < screenWidth; j++){w_arr[j] = '=';}int p = currentWidth > screenWidth ? currentWidth % screenWidth : currentWidth;w_arr[p - 1] = '>';if (p < screenWidth)w_arr[p] = '>';if (p+1 < screenWidth)w_arr[p+1] = ' ';string w_str = p >= screenWidth ? "" : new string(w_arr,0,p+1);double dpCount = writeIndex * 100 / dcCount;Console.ForegroundColor = ConsoleColor.Green;Console.Write(w_str);Console.ForegroundColor= ConsoleColor.White;string w_str2=p>= screenWidth? "": new string(w_arr,p+1,screenWidth-p-1);Console.Write("{0}>[{1}/{2}]({3}%)", w_str2, readIndex, recordCount, dpCount.ToString("f2"));}currentWidth = currentWidth <= screenWidth ? currentWidth + 1 : 1;
}

View Code

下面是模拟显示进度的代码调用方法：

int recordCount = 3721;
Console.WriteLine("【模拟】开始处理数据：");
ConsoleProcessDisplayer displayer = new ConsoleProcessDisplayer(recordCount);
Console.WriteLine("插入记录数：");
displayer.Begin();
for (int i = 0;i< recordCount; i++)
{displayer.DisplayProcessing(i);System.Threading.Thread.Sleep(10);
}
displayer.DisplayProcessing(recordCount);
displayer.End();