如何跨库迁移一张大表数据

SmileNicky

发布于 2025-10-14 08:44:49

10900

代码可运行

文章被收录于专栏：Nicky's blogNicky's blog

运行总次数：0

代码可运行

跨库迁移一张大表数据可以采用以下几种方法：

使用数据库自带工具
- MySQL：可以使用mysqldump工具。先导出数据，若要导出完整表数据，命令为mysqldump -umyuser -pmy_passwd --single - transaction src_db big_table > src_db.big_table.sql；若要导出部分数据（不导出建表语句），命令为mysqldump -umyuser -pmy_passwd -t src_db big_table --single - transaction --where="条件" > big_table - 部分数据.sql。然后在目标数据库中使用mysql -umyuser -pmy_passwd dst_db < big_table.sql导入数据。
- Oracle：可以使用EXPDP和IMPDP工具。先在源数据库上使用EXPDP命令将大表数据导出为.dmp文件，如EXPDP user/password@source_db schemas=source_schema tables=big_table directory=dp_dir dumpfile=big_table.dmp，然后在目标数据库上使用IMPDP命令将.dmp文件中的数据导入，如IMPDP user/password@target_db directory=dp_dir dumpfile=big_table.dmp。
利用ETL工具：以Pentaho Data Integration（Kettle）为例，首先创建一个新的转换，添加源数据库连接和表输入步骤，配置源数据库连接信息和要迁移的表；接着添加目标数据库连接和表输出步骤，配置目标数据库连接信息和目标表；然后通过映射字段，将源表字段与目标表字段对应起来；最后运行转换，实现数据迁移。
通过编程实现：以Python为例，使用pymysql库可以实现MySQL数据库之间的大表数据迁移。示例代码如下：

import pymysql
import time

def migrate_data():
    # 连接源数据库
    source_conn = pymysql.connect(host='source_host', user='user', password='password', database='source_db')
    source_cursor = source_conn.cursor()
    # 连接目标数据库
    target_conn = pymysql.connect(host='target_host', user='user', password='password', database='target_db')
    target_cursor = target_conn.cursor()
    # 假设按照主键区间进行分批迁移，每次迁移10000条数据
    batch_size = 10000
    start_id = 0
    total_records = 0
    while True:
        start_time = time.time()
        # 从源数据库查询数据
        source_cursor.execute(f"SELECT * FROM large_table WHERE id >= {start_id} LIMIT {batch_size}")
        data = source_cursor.fetchall()
        if not data:
            break
        # 将数据插入到目标数据库
        for row in data:
            try:
                target_cursor.execute("INSERT INTO target_table (column1, column2,...) VALUES (%s, %s,...)", row)
                total_records += 1
            except pymysql.IntegrityError as e:
                # 处理主键冲突，这里假设选择跳过
                if "Duplicate entry" in str(e):
                    continue
                else:
                    raise e
        # 更新起始主键值
        start_id += batch_size
        end_time = time.time()
        print(f"已迁移批次：{(start_id // batch_size)}, 已迁移记录数：{total_records}, 耗时：{end_time - start_time}秒")
    # 提交事务并关闭连接
    target_conn.commit()
    source_conn.close()
    target_conn.close()


# 调用函数开始迁移
migrate_data()

在跨库迁移大表数据时，还需要注意以下事项：