mysql迁移到mongodb_从mysql迁移数据至mongoDB

简述

最近做的积分系统二期要上线了，在第一期的时候仅考虑到了mysql分表，未引入mongoDB，因而在在第二期的时候引入Mongo，除了程序中要多写(同时写mysql,Mongodb)之外，还需要考虑到历史数据的迁移(拷贝)问题。

方案制定

经查发现：线上mysql数据问题接近200W，平均每日新增3~4W条。压力和数据量都不算大，所以决定直接用脚本进行迁移。

因为线上服务不会停，所以在迁移数据时，系统仍会不断有新增数据，那么如果处理这部分的新增数据也是一个问题。

针对这个问题解决办法比较多，但是主要思想都是数据分段。比如：

以时间维度分段，以某个时间为截点timeA，先迁移这部分数据，然后系统上后再导入timeA至上线时这部分的数据，这个办法有个弊端，时间精准性很难把握，你很难记录系统上线的时间timeB.

当然有个办法就是等系统上线之后，去mongoDB中查询第一条记录插入的时间，以此做为timeB的值，但是这个做法仍不是比较好的解决办法。如果QPS很低，迟迟没有记录插入，那么你只能等，如果QPS过高，则可能同一秒的时候有些数据仍没有写进mongoDB，只写入了mysql，以此时间为timeB的值，则可能会漏掉记录。

以唯一键分段比如id，比如导出数据时记录每个表导出记录的id值。然后采用跟上面的思路一样，进行2次分批导入，缺点仍然是麻烦。

所以针对这个问题，结合我们业务访问量以及数据量制定的方案就是：先重叠数据，再去重。即让线上的数据和迁移的数据产生一部分重叠，再根据唯一性条件，删除重复数据,大致思路如下：

在线上mongoDB建立一个临时库tmpDB

发布新系统，系统中使用的mongoDB先用临时库tmpDB，保证线上数据在多写时写入tmpDB

从线上mysql导出数据到csv.

将导出的csv导入至mongoDB的正式库

将线上系统切至正式库

将tmpDB的数据导入正式库

在正式库中数据去重复

注意： 第2步和第3步一定不能不能颠倒顺序，否则就不是数据重复，而是数据遗漏。

方案实施

1. 将数据从mysql导出至csv文件

此步比较简单，直接select .. from ..into outfile ...，大致代码如下：

select userId,username,businessType,orderId,handlePoint,

balance,concat(createTime,'.000Z') as createTime,clientType,

'com.project.points.mongoModel.PointChangeDetail'

from t_point_record_0

union all

select userId,username,businessType,orderId,handlePoint,

balance,concat(createTime,'.000Z') as createTime,clientType,

'com.project.points.mongoModel.PointChangeDetail'

from t_point_record_1

into outfile '/home/tmp/point_record.csv'

fields terminated by ','optionally enclosed by ''lines terminated by '\n';

当然实际环境上我不是分了2张表，而是10张，为节省篇幅，没必要完整列出来，此处有几个值得注意的点(坑)：

1.1 createTime在mysql中是日期类型，在导出时我们将他变成了varchar(String)类型。

此处是一个大坑，mysql的日期类型在导出之后csv之后，无论是什么格式的日期类型，亦或是timestamp，在导入mongoDB时都不能正确识别为日期类型。

所以我在观察了程序插入mongoDB的数据格式后，干脆将他导出为String格式，并用concat(createTime,'.000Z')这样的语法进行拼装，以满足mongoDB的格式要求。

1.2 因为我采用的是spring-data-mongo，在插入数据的时默认会插入Object的Class，故而我在导出数据时将Class写死以便导入。

1.3 从mysql导入到MongoDB的时候，基本上就可以放弃id字段了。

2.导入数据

直接采用mongoimport进行导入，具体参数可以参见官方文档，有比较详细的解释。我们此处的用法如下：

mongoimport -u 用户名 -p 密码 -d db名 -c collection名

--type csv --ignoreBlanks --fields userId,username,businessType,orderId,

handlePoint,balance,createTime,clientType,_class

--file 'point_record.csv'

这里仍然有几个要注意的地方：

注意属性与csv中列的对应顺序

属性之间用,隔开，中间__千万不要有空格__

3. 进行日期转换

之前说过了mysql导出的日期格式是String，在导入mongoDB以后，发现仍然是String，因此需要执行转换函数。

db.pointChangeDetail.find().forEach( function(obj) {

obj.createTime= new ISODate(obj.createTime);

db.pointChangeDetail.save(obj);

});

这个函数执行略有点耗时。

4. 从tmpDB导入正式库

这步基本上就不说了

5.去重

db.pointChangeDetail.aggregate([

{ $group: {

_id: {

orderId: "$orderId"

},

dups: { "$addToSet": "$_id" },

count: { "$sum": 1 }

}},

{ $match: {

count: { "$gt": 1 }

}}

],

{

allowDiskUse: true

}).result.forEach(function(doc) {

doc.dups.shift();

db.signDetail.remove({_id : {$in: doc.dups }});

})

如果以上代码不能工作，则使用如下代码：

db.runCommand(

{ aggregate: "pointChangeDetail",

pipeline: [

{$group: {_id: {orderId: "$orderId"}, dups: { "$addToSet": "$_id" }, cnt: {$sum: 1}}},

{$match: {cnt: { "$gt": 1 }}}

],

allowDiskUse: true

}

).result.forEach(function(doc) {

doc.dups.shift();

db.pointChangeDetail.remove({_id : {$in: doc.dups }});

})

本次数据迁移工作就算完成了。