如何快速将MySQL库A的data表迁移至结构不同的库B的reading表

问题背景 #

需将MySQL数据库A中data表的所有数据迁移至数据库B的reading表，两表结构异构无法直接使用mysqldump。当前采用的批量插入存储过程可正常运行，但耗时已超一天，代码如下：

CREATE `uspInsertIgnoreDataInBatches`()
BEGIN
    DECLARE start_index INT DEFAULT 0;
    DECLARE batch_size INT DEFAULT 100000;
    DECLARE total_rows INT;

    -- 获取总行数
    SELECT COUNT(*) INTO total_rows
    FROM A.data
    LEFT JOIN A.device_sensor ON data.device_sensor_id = device_sensor.id
    LEFT JOIN A.sensors ON sensors.id = device_sensor.sensor_id;

    -- 循环处理所有批次
    WHILE start_index < total_rows DO
        BEGIN
            -- 批量插入数据
            INSERT IGNORE INTO B.reading (Id, RecordedAtMT, SensorId, LocationId, SensorValue)
            SELECT data.id, data.datetime, sensors.id as sensor_id, data.location_id, data.value
            FROM A.data
            LEFT JOIN A.device_sensor ON data.device_sensor_id = device_sensor.id
            LEFT JOIN A.sensors ON sensors.id = device_sensor.sensor_id
            ORDER BY A.data.datetime DESC
            LIMIT batch_size OFFSET start_index;
        END;

        -- 移动到下一批次
        SET start_index = start_index + batch_size;
    END WHILE;
END

注：使用INSERT IGNORE是为了处理源表部分数据违反目标表唯一约束的情况。

现有代码性能瓶颈分析 #

1. OFFSET分页性能衰退：随着start_index增大，LIMIT ... OFFSET需要扫描越来越多的行后再丢弃，后续批次的查询速度会急剧下降。
2. 预查询COUNT(*)的额外开销：全表JOIN后统计行数的操作本身耗时极长，若迁移过程中源表数据有变化，统计结果也会失效。
3. INSERT IGNORE的约束检查开销：每一行都要经过唯一约束校验，冲突时仅忽略，相比提前过滤的方式效率更低。
4. 重复JOIN计算：每个批次都要重复执行三次表JOIN，没有复用中间结果，浪费计算资源。

具体优化方案 #

1. 用主键范围扫描替代OFFSET分页 #

利用源表主键（如data.id）的有序性进行分段查询，避免OFFSET的全表扫描开销，批次速度保持稳定：

CREATE PROCEDURE `uspInsertIgnoreDataInBatches`()
BEGIN
    DECLARE last_id BIGINT DEFAULT 0;
    DECLARE batch_size INT DEFAULT 100000;
    DECLARE row_count INT DEFAULT 1;

    WHILE row_count > 0 DO
        BEGIN
            INSERT IGNORE INTO B.reading (Id, RecordedAtMT, SensorId, LocationId, SensorValue)
            SELECT 
                d.id, d.datetime, s.id as sensor_id, d.location_id, d.value
            FROM A.data d
            LEFT JOIN A.device_sensor ds ON d.device_sensor_id = ds.id
            LEFT JOIN A.sensors s ON s.id = ds.sensor_id
            WHERE d.id > last_id
            ORDER BY d.id ASC
            LIMIT batch_size;

            SET row_count = ROW_COUNT();
            -- 获取当前批次的最大主键，作为下一批的起始条件
            SELECT MAX(d.id) INTO last_id FROM (
                SELECT d.id FROM A.data d
                WHERE d.id > last_id
                ORDER BY d.id ASC
                LIMIT batch_size
            ) AS temp;
        END;
    END WHILE;
END

若data.datetime是唯一且有序字段，也可作为分段依据，但主键通常是最优选择。

2. 移除预查询COUNT(*)，用ROW_COUNT()判断循环结束 #

原代码的COUNT(*)操作会占用大量时间，改用ROW_COUNT()获取每次INSERT的实际行数，当行数为0时说明无更多数据，直接结束循环，避免不必要的预查询。

3. 优化冲突处理逻辑，减少约束检查 #

若唯一约束针对Id字段，可提前在源端过滤目标表已存在的记录，减少目标端的约束校验次数：

-- 在数据库B创建临时表，存储已存在的Id
CREATE TEMPORARY TABLE existing_ids (id BIGINT PRIMARY KEY);
INSERT INTO existing_ids SELECT Id FROM B.reading;

-- 迁移时过滤已存在的记录
CREATE PROCEDURE `uspInsertFilteredData`()
BEGIN
    DECLARE last_id BIGINT DEFAULT 0;
    DECLARE batch_size INT DEFAULT 100000;
    DECLARE row_count INT DEFAULT 1;

    WHILE row_count > 0 DO
        BEGIN
            INSERT INTO B.reading (Id, RecordedAtMT, SensorId, LocationId, SensorValue)
            SELECT 
                d.id, d.datetime, s.id as sensor_id, d.location_id, d.value
            FROM A.data d
            LEFT JOIN A.device_sensor ds ON d.device_sensor_id = ds.id
            LEFT JOIN A.sensors s ON s.id = ds.sensor_id
            LEFT JOIN existing_ids ei ON d.id = ei.id
            WHERE d.id > last_id AND ei.id IS NULL
            ORDER BY d.id ASC
            LIMIT batch_size;

            SET row_count = ROW_COUNT();
            SELECT MAX(d.id) INTO last_id FROM (
                SELECT d.id FROM A.data d
                WHERE d.id > last_id
                ORDER BY d.id ASC
                LIMIT batch_size
            ) AS temp;
        END;
    END WHILE;
END

临时表需添加主键索引，保证JOIN过滤的效率。此方案可完全避免INSERT IGNORE的约束检查开销。

4. 调整数据库参数，降低写入开销 #

• 禁用目标表非必要索引：迁移前禁用reading表除主键/唯一键外的所有索引，迁移完成后重建，避免每次插入更新索引的开销：

  ALTER TABLE B.reading DISABLE KEYS;
  -- 执行迁移操作
  ALTER TABLE B.reading ENABLE KEYS;
  

• 关闭自动提交：在存储过程开头添加SET autocommit = 0;，结尾添加COMMIT;，减少事务日志的刷盘次数。
• 调大InnoDB缓冲池：确保目标数据库的innodb_buffer_pool_size足够大，尽可能将写入数据留在内存中，减少磁盘IO。

5. 预生成中间表，复用JOIN结果 #

若源表数据在迁移过程中不会变化，可先在数据库A中生成包含所有目标字段的临时表，避免每个批次重复执行JOIN：

-- 在数据库A创建带主键的临时表，预计算JOIN结果
CREATE TEMPORARY TABLE temp_migration_data (
    id BIGINT,
    datetime DATETIME,
    sensor_id BIGINT,
    location_id BIGINT,
    value DECIMAL(10,2),
    PRIMARY KEY (id)
) ENGINE=InnoDB;

INSERT INTO temp_migration_data
SELECT 
    d.id, d.datetime, s.id as sensor_id, d.location_id, d.value
FROM A.data d
LEFT JOIN A.device_sensor ds ON d.device_sensor_id = ds.id
LEFT JOIN A.sensors s ON s.id = ds.sensor_id;

-- 从临时表迁移至目标表
CREATE PROCEDURE `uspInsertFromTemp`()
BEGIN
    DECLARE last_id BIGINT DEFAULT 0;
    DECLARE batch_size INT DEFAULT 100000;
    DECLARE row_count INT DEFAULT 1;

    WHILE row_count > 0 DO
        BEGIN
            INSERT IGNORE INTO B.reading (Id, RecordedAtMT, SensorId, LocationId, SensorValue)
            SELECT id, datetime, sensor_id, location_id, value
            FROM A.temp_migration_data
            WHERE id > last_id
            ORDER BY id ASC
            LIMIT batch_size;

            SET row_count = ROW_COUNT();
            SELECT MAX(id) INTO last_id FROM (
                SELECT id FROM A.temp_migration_data
                WHERE id > last_id
                ORDER BY id ASC
                LIMIT batch_size
            ) AS temp;
        END;
    END WHILE;
END

此方案将JOIN计算一次性完成，后续迁移仅需读取临时表，大幅减少重复计算开销。

内容的提问来源于stack exchange，提问作者kj49