快捷導航

MySql使用存儲過程進行單表數(shù)據(jù)遷移的實現(xiàn)

更新時間：2023年11月10日 10:06:33 作者：生命猿于運動

近期在進行業(yè)務(wù)解耦，對冗余在一起切又屬于不同業(yè)務(wù)的代碼進行分離，同時也將數(shù)據(jù)庫進行分離存儲，那么這時候就涉及到多個表的數(shù)據(jù)要進行遷移，本文就來介紹一下MySql使用存儲過程進行單表數(shù)據(jù)遷移，感興趣的可以了解一下

前景

近期在進行業(yè)務(wù)解耦，對冗余在一起切又屬于不同業(yè)務(wù)的代碼進行分離，同時也將數(shù)據(jù)庫進行分離存儲，那么這時候就涉及到多個表的數(shù)據(jù)要進行遷移，這里我們就來總結(jié)一下如何使用存儲過程進行數(shù)據(jù)高效遷移。

設(shè)計思路

跨數(shù)據(jù)庫實例表數(shù)據(jù)遷移，無非就是把一個表完完整整的復制到另一個數(shù)據(jù)庫實例當中，但是怎么做才能簡單易懂又高效呢？

首先我們寫一個腳本肯定也希望能夠多次使用，否則何必浪費時間去研究寫大批量處理的腳本呢！我們先分析數(shù)據(jù)遷移的一些主要步驟：目標實例創(chuàng)建表、數(shù)據(jù)分批處理、數(shù)據(jù)遷移記錄、數(shù)據(jù)遷移入庫

目標實例創(chuàng)建表：我們需要根據(jù)個人所需，明確是否強制重新創(chuàng)建表，通常情況既然是全表遷移那都是要強制重新建表的（無論是否已有表數(shù)據(jù)）。
數(shù)據(jù)分批處理：要對數(shù)據(jù)進行分批處理，首先我們需要對數(shù)據(jù)進行排序，那么排序最好我們是以自增主鍵id進行排序，這樣方便我們進行分批數(shù)據(jù)獲取。
數(shù)據(jù)遷移記錄：這里我們最好有個臨時表用來做實時數(shù)據(jù)遷移記錄，以免大數(shù)據(jù)遷移我們都不知道遷移到哪了，同時臨時表也有助于數(shù)據(jù)分批處理。
數(shù)據(jù)遷移入庫：對數(shù)據(jù)進行排好序后，我們根據(jù)主鍵id對數(shù)據(jù)進行過濾，避免在limit后面進行分頁操作，limit只需要確認遷移數(shù)據(jù)量即可。

下面我們就具體分析一下完整的遷移腳本。

遷移腳本

首先我們需要明確數(shù)據(jù)要遷移的目標數(shù)據(jù)庫，最好要把這點寫在腳本里，方式跑錯數(shù)據(jù)庫實例：

USE `test_db`;

臨時表創(chuàng)建：主要輔助記錄需要遷移的數(shù)據(jù)量、實時更新已遷移的數(shù)據(jù)量、遷移表名、遷移的數(shù)據(jù)最大主鍵id用來進行數(shù)據(jù)過濾。

DROP TABLE IF EXISTS tmp_migrate_table_record;
CREATE TABLE tmp_migrate_table_record(
    `id` INT PRIMARY KEY NOT NULL AUTO_INCREMENT COMMENT '主鍵ID',
    `table_name` VARCHAR(50) COMMENT '遷移表名',
    `source_table_count` INT COMMENT '源表記錄數(shù)',
    `target_table_count` INT COMMENT '源表記錄數(shù)',
    `max_id` BIGINT COMMENT '已遷移最大主鍵id'
);

全表遷移存儲過程實現(xiàn)，主要包含以下幾個參數(shù)：

sourceSchema：數(shù)據(jù)源schema名稱
tableName：需要遷移的表名（若需要遷移的目標表名與源表名不一致，可自行添加參數(shù)做修改）
primaryKey：表主鍵名，用于數(shù)據(jù)排序、過濾、分批處理
batchSize：大數(shù)據(jù)遷移分批次大小

DROP PROCEDURE IF EXISTS p_migrate_full_table_data;
DELIMITER $$
CREATE PROCEDURE p_migrate_full_table_data(IN sourceSchema VARCHAR(50), IN tableName VARCHAR(50), IN primaryKey VARCHAR(50), IN batchSize INT)
BEGIN    
    -- 判斷舊表是否存在，若存在則刪除舊表（強制重新建表）
    SET @dropTableSql = CONCAT('DROP TABLE IF EXISTS ', tableName, ';');
    PREPARE dropTableSql FROM @dropTableSql;
    EXECUTE dropTableSql;
    
    -- 依賴源數(shù)據(jù)，創(chuàng)建新表（這里可以根據(jù)需要更換表名）
    SET @craeteTable = CONCAT('CREATE TABLE ', tableName, ' LIKE `', sourceSchema, '`.', tableName, ';');
    PREPARE craeteTable FROM @craeteTable;
    EXECUTE craeteTable;

    -- 清除當前表遷移數(shù)據(jù)記錄數(shù)，防止有舊數(shù)據(jù)影響
    SET @deleteCountSql = CONCAT('DELETE FROM tmp_migrate_table_record WHERE table_name=''', tableName, ''';');
    PREPARE deleteCountSql FROM @deleteCountSql;
    EXECUTE deleteCountSql;
    
    -- 初始記錄臨時表正在進行數(shù)據(jù)遷移的表信息
    SET @countSql = CONCAT('INSERT INTO tmp_migrate_table_record(table_name, source_table_count, target_table_count) SELECT ''', tableName, ''', COUNT(*), 0 FROM `', sourceSchema, '`.', tableName, ';');
    PREPARE countSql FROM @countSql;
    EXECUTE countSql;
    
    -- 用于查看預(yù)編譯后的SQL腳本，若有需要可以打開注釋查看
    -- SELECT @dropTableSql, @craeteTable, @deleteCountSql, @countSql;
    
    -- 數(shù)據(jù)源表記錄數(shù)
    SET @sourceCount = 0;
    -- 目標表記錄數(shù)
    SET @targetCount = 0;
    -- 已導入最大id
    SET @maxId = 0;
    
    SELECT source_table_count, target_table_count, IFNULL(max_id, 0) INTO @sourceCount, @targetCount, @maxId FROM tmp_migrate_table_record WHERE table_name=tableName;
    
    -- 循環(huán)分批遷移數(shù)據(jù)，根據(jù)已遷移數(shù)量與需要遷移數(shù)量進行比較
    WHILE @sourceCount <> @targetCount DO
        -- 開啟事務(wù)
        START TRANSACTION;
        
        -- 執(zhí)行數(shù)據(jù)分批遷移
        SET @insertSql = CONCAT('INSERT INTO ', tableName, ' SELECT * FROM `', sourceSchema, '`.', tableName, ' WHERE ', primaryKey, ' > ', @maxId, ' ORDER BY ', primaryKey, ' ASC ', 'LIMIT ', batchSize, ';');
        PREPARE insertSql FROM @insertSql;
        EXECUTE insertSql;
        
        -- 實時更新臨時表已遷移記錄數(shù)
        SET @updateCountSql = CONCAT('UPDATE tmp_migrate_table_record SET target_table_count = (SELECT COUNT(*) FROM ', tableName, ') WHERE table_name=''', tableName, ''';');
        PREPARE updateCountSql FROM @updateCountSql;
        EXECUTE updateCountSql;
        
        -- 跟新臨時表以遷移數(shù)據(jù)最大主鍵id
        SET @updateMaxIdSql = CONCAT('UPDATE tmp_migrate_table_record SET max_id = (SELECT IFNULL(MAX(', primaryKey,'), 0) FROM ', tableName, ') WHERE table_name=''', tableName, ''';');
        PREPARE updateMaxIdSql FROM @updateMaxIdSql;
        EXECUTE updateMaxIdSql;
        
        -- 刷新變量
        SELECT target_table_count, max_id INTO @targetCount, @maxId FROM tmp_migrate_table_record WHERE table_name=tableName;
        
        -- 查看上面拼裝后的sql，需要排查問題可打開查看
        SELECT @insertSql, @updateCountSql, @updateMaxIdSql, @sourceCount, @targetCount;
        
        -- 提交事務(wù)
        COMMIT;
    END WHILE;
END $$
DELIMITER ;

存儲過程完成后，接下來就是執(zhí)行存儲過程進行數(shù)據(jù)遷移了：

CALL p_migrate_full_table_data('test_source_db', 't_student', 'student_id', 50000);
CALL p_migrate_full_table_data('test_source_db', 't_student', 'student_id', 50000);
CALL p_migrate_full_table_data('test_source_db', 't_student', 'student_id', 50000);

確認數(shù)據(jù)沒有問題后，可以對臨時表進行清理：

DROP TABLE IF EXISTS tmp_migrate_table_record;

遷移示例

創(chuàng)建測試表

DROP TABLE IF EXISTS t_student;
CREATE TABLE t_student(
    id BIGINT,
    `name` VARCHAR(50),
    age INT(3),
    state CHAR(1),
    PRIMARY KEY (id)
);

DROP TABLE IF EXISTS t_course;
CREATE TABLE t_course(
    id BIGINT,
    `name` VARCHAR(50)
);

使用存儲過程初始化數(shù)據(jù)

DROP PROCEDURE IF EXISTS init_student;
DELIMITER $$
CREATE PROCEDURE init_student()
BEGIN
    DELETE FROM t_student;
    SET @p = 1;
        -- 測試數(shù)據(jù)數(shù)量自己定
    WHILE @p < 234567 DO
        INSERT INTO t_student
        VALUES(@p, CONCAT('user', @p * 1000000), 18, 'A');
        SET @p = @p + 1;
    END WHILE;
END $$
DELIMITER ;

CALL init_student();

執(zhí)行遷移腳本

CALL p_migrate_full_table_data('test_source_db', 't_student', 'id', 50000);
CALL p_migrate_full_table_data('test_source_db', 't_course', 'id', 50000);