MySQL表数据迁移自动化

分享到

MySQL表数据迁移自动化

发布于2013-3-19

一、背景

之前我写过关于SQL Server的数据迁移自动化的文章：SQL Server 数据库迁移偏方，在上篇文章中设计了一张临时表，这个临时表记录搬迁的配置信息，用一个存储过程读取这张表进行数据的迁移，再由一个Job进行迭代调用这个存储过程。

在这次MySQL的实战中，我的数据库已经做了4个分片，分布在不同的4台机器上，每台机器上的数据量有1.7亿（1.7*4=6.8亿），占用空间260G（260*4=1040G），这次迁移的目的就是删除掉一些历史记录，减轻数据库压力，有人说这为什么不使用表分区呢？这跟我们的业务逻辑有关造成无法使用表分区，至于为什么，参考阅读：MySQL表分区实战，其中最重要就是唯一索引的问题，扩展阅读：MySQL当批量插入遇上唯一索引，这篇文章需要了解MySQL的定时器的一些知识：MySQL定时器Events

本文与SQL Server 数据库迁移偏方最大的不同就是MySQL的Events不是串行执行的，当作业调用的存储过程还没有执行完毕，但又到了调度的时间，MySQL不会等待上次作业完成之后再调度，所以会造成重复调用读取到相同的数据；而SQL Server并不存在上面的问题。

二、设计思路

1. 创建一个临时表TempBlog_Log，这个表用于保存每次转移数据的ID起始值和结束值，以及搬迁的开始时间和结束时间；（这个ID是我们要迁移表的主键，自增字段，唯一标识）

2. 创建一个存储过程InsertData()，这个存储过程用于在TempBlog_Log表中插入记录，创建这个存储过程是因为MySQL跟SQL Server有些不同，MySQL不支持匿名存储过程，SQL Server直接执行SQL就可以了，无需为这些SQL再创建一个存储过程，这就是匿名存储过程了；

3. 创建一个存储过程MoveBlogData()，这个存储过程用于在TempBlog_Log表中读取记录，再批量把BlogA数据转移到BlogB中；这个是核心逻辑，解决了定时器重复调度的问题，详情见代码的解释；

4. 创建一个定时器e_Blog, 这个定时器定时调用存储过程MoveBlogData()，但是这里存在重复调度的问题，只能通过存储过程MoveBlogData()进行控制。

三、迁移自动化特点

1. 该设计适应于大数据的迁移；

2. 可以最小化宕机时间（在转移的过程中BlogA还是一直在进数据的，只是在最后一部分数据的时候需要短时间的停入库操作）；

3. 可以防止MySQL定时器重复执行所带来的问题；

4. 可以实时监控数据转移的进度；

5. 数据迁移可能需要持续好几天的时间，它能保证BlogB的数据会无限的接近BlogA的数据；

四、代码分析

(一) 创建临时表TempBlog_Log

-- 创建表
CREATE TABLE TempBlog_Log(
    BeginId INT NOT NULL,
    EndId INT NOT NULL,
    IsDone BIT DEFAULT b'0' NOT NULL,
    BeginTime DATETIME DEFAULT NULL,
    EndTime DATETIME DEFAULT NULL,
PRIMARY KEY(BeginId) 
);

下面就对表结构进行字段解释：

1) BeginId、EndId都是ServerA迁移表的主键值，BeginId表示一次数据迁移的起始值，EndId表示一次数据迁移的结束值，两个值的差就是这次数据转移的数据量；

2) IsDone 表示是否已经成功转移数据；

3) BeginTime表示转移的开始时间，EndTime表示转移的结束时间，这两个字段设置缺省值为NULL很关键，是后面进行判断是否重复执行的依据；

(二) 创建存储过程InsertData()

-- 存储过程
DELIMITER $$
USE `DataBaseName`$$
DROP PROCEDURE IF EXISTS `InsertData`$$

CREATE DEFINER=`root`@`%` PROCEDURE `InsertData`()
BEGIN
    DECLARE ids_begin,ids_end,ids_increment INT;
    SET ids_begin=130000000;-- 需要转移开始Id值
    SET ids_end=210000000;-- 需要转移结束Id值
    SET ids_increment=200000;-- 每次转移的Id量
    WHILE ids_begin < ids_end DO 
        INSERT INTO TempBlog_Log(BeginId,EndId) VALUES(ids_begin,ids_begin+ids_increment);
        SET ids_begin = ids_begin + ids_increment;
    END WHILE; 
END$$
    DELIMITER ;

MySQL中不支持匿名存储过程，所以为了在临时表TempBlog_Log插入记录，只能创建一个存储过程了，如果你还没写过MySQL的存储过程，那么这是一个很好的例子。

1) 为了能在存储过程中使用MySQL的分隔符“;”，DELIMITER $$表示你以“$$”作为分隔符，你也可以使用“//”；

2) 定义变量时，你需要把所有的变量定义完了，之后再进行赋值，不然会报错，这跟SQL Server是有区别的；

3) WHILE条件后面需要加DO，而且要以END WHILE;作为结束标记；

4) 作为存储过程的结束，再次出现“$$”表示已经结束，跟上一个“$$”形成一个整体、过程，并重新设置“;”为分隔符；

5) 执行CALL InsertData();调用上面的存储过程，插入数据，调用完毕的结果如下图Figure1所示

（Figure1：转移前状态）

(三) 创建保留数据的新表BlogB

做完上面的准备工作，接下来就是创建与BlogA相同结构的BlogB表了，有些不同的就是不需要在BlogB创建太多的索引，只需要存储两个索引就可以了，一个是ID的聚集索引，一个是唯一索引（在批量插入的时候需要判重）；

上面索引是根据我业务上的需求决定的，你需要视情况而定；

(四) 创建存储过程MoveBlogData()

DELIMITER $$
USE `DataBaseName`$$
DROP PROCEDURE IF EXISTS `MoveBlogData`$$
CREATE DEFINER=`root`@`localhost` PROCEDURE `MoveBlogData`()
BEGIN
    DECLARE blog_ids_begin INT;-- Id起始值
    DECLARE blog_ids_end INT;-- Id结束值
    DECLARE blog_ids_max INT;-- BlogA表现在的最大值
    DECLARE blog_begintime INT;-- 执行开始时间
    DECLARE blog_endtime INT;-- 执行结束时间
    -- 查询TempBlog_Log表还没有done的记录
    SELECT BeginId,EndId,BeginTime,EndTime INTO blog_ids_begin,blog_ids_end,blog_begintime,blog_endtime 
FROM TempBlog_Log WHERE IsDone = 0 ORDER BY BeginId LIMIT 0,1;
     -- 防止了定时器的重复执行
    IF(blog_begintime IS NULL AND blog_endtime IS NULL) THEN
        -- 设置当前最大的Id值
        SELECT MAX(ids) INTO blog_ids_max FROM BlogA;
        -- 防止转移超过当前最大值的Id数据
        IF(blog_ids_begin != 0 AND blog_ids_end != 0 AND blog_ids_max >= blog_ids_end) THEN
            -- 更新执行开始时间
            UPDATE TempBlog_Log SET BeginTime = NOW() WHERE BeginId = blog_ids_begin;
            -- 插入Id段数据，忽略重复值
            INSERT IGNORE INTO BlogB (ID,AuthorID,Content,QUOTE,QuoteID,Author,TIME,Url,ImageUrl,Transmits
,Comments,HASH,Site,AuthorUID,TYPE,HotTopic,AddOn,QuoteAuthorID,IDs)
            SELECT ID,AuthorID,Content,QUOTE,QuoteID,Author,TIME,Url,ImageUrl,Transmits,Comments,
HASH,Site,AuthorUID,TYPE,HotTopic,AddOn,QuoteAuthorID,IDs
                FROM BlogA WHERE IDs >= blog_ids_begin AND IDs < blog_ids_end;
            -- 更新执行结束时间
            UPDATE TempBlog_Log SET IsDone = 1,EndTime = NOW() WHERE BeginId = blog_ids_begin;
        END IF;
    END IF;
END$$
DELIMITER ;

这个存储过程是整个搬迁数据的核心代码，之所以说是核心，是因为它把比较多的细节考虑进去，基本上实现自动化的目的。

1) 代码中IF(blog_begintime IS NULL AND blog_endtime IS NULL) 防止了定时器的重复执行，两个值都为NULL的时候表示这个Id段的数据还没有被转移，这样就可以跳过，不执行下面的逻辑；

2) 查询BlogA的最大值可以防止转移超过当前BlogA最大值的Id数据，只有当blog_ids_max>=blog_ids_end才符合转移的条件；

3) 在MySQL中对唯一索引约束的数据操作有很多的关键字支持，INSERT IGNORE INTO就是在批量插入过程中只插入没有的数据，忽略重复的数据；更多唯一索引的信息：MySQL当批量插入遇上唯一索引

4) 查询中FROM BlogA WHERE IDs >= blog_ids_begin AND IDs < blog_ids_end;需要注意IDs值的闭合关系，不然造成重复数据或者丢失数据；

(五) 创建定时器e_Blog

DELIMITER $$

CREATE DEFINER=`root`@`localhost` EVENT `e_blog` 
ON SCHEDULE EVERY 30 SECOND 
STARTS '2012-12-07 14:58:53' 
ON COMPLETION PRESERVE DISABLE 
DO CALL MoveBlogData()$$

DELIMITER ;

这定时器e_Blog的作用是在每隔30 SECOND调用一次存储过程MoveBlogData()，至于有没转移数据那就是存储过程判断了，跟定时器的调度频率完全没有关系，更多关于定时器的信息：MySQL定时器Events

(六) 监控数据转移的状态

当定时器启动后，可以查看TempBlog_Log表监控调度的进度：

（Figure2：转移中状态）

Figure2表示正在转移Id>=225200000到Id<225400000这20W的数据；

你也可以通过下面的SQL进行统计：

SELECT IsDone,COUNT(1) FROM tempblog_log 
GROUP BY IsDone ORDER BY IsDone DESC;

(七) 创建索引

创建保留数据的新表BlogB的时候不要创建不必要的索引，等转移完数据之后再创建回相关的索引；这样做的目的是在插入数据的时候不需要对索引进行维护，并且到转移完之后再创建索引可以让索引更加没有索引碎片；

(八) 禁用定时器

当TempBlog_Log表不再更新的时候，我们就可以禁用定时器了。因为BlogA表是一直在进数据的，所以当TempBlog_Log不再更新就说明数据已经基本转移完毕了（新增的数据量小于20W），这个时候就可以禁用定时器了。

(九) 转移最后数据

首先停止对BlogA表的入库操作，通过SQL转移最后一部分的数据到BlogB中，转移完之后修改表名就大功告成了。

五、参考文献

SQL Server 数据库迁移偏方

MYSQL插入处理重复键值的几种方法