怎样用脚本批量迁移数据库？

wen 实用脚本 2026-07-22 1

用Python脚本实现零停机迁移

怎样用脚本批量迁移数据库？

目录导读

为什么要用脚本批量迁移数据库？——痛点与场景分析
脚本迁移的核心技术栈选型对比
从零搭建批量迁移脚本（含完整代码示例）
关键问题问答：迁移中数据一致性、性能与回滚策略
生产环境迁移的最佳实践与避坑指南

为什么要用脚本批量迁移数据库？

在日常运维中,数据库迁移是高频操作，无论是从MySQL迁移到PostgreSQL，还是将SQL Server数据迁移到云原生数据库（如TiDB或GaussDB），手动操作不仅耗时，还可能因人为失误导致数据丢失，脚本批量迁移的核心优势在于：

自动化：一次编写，多次执行，支持定时任务或CI/CD触发。
可重复性：迁移逻辑固化在代码中，避免“记错表名”或“漏掉字段”。
可审计：每一条数据的迁移状态都可以记录日志，便于事后追溯。
零停机：通过增量同步策略，业务侧几乎无感知。

适用场景：数据库版本升级、跨云迁移、分库分表重构、同构/异构数据库复制。

脚本迁移的核心技术栈选型

方案	适用场景	语言/工具	优缺点
使用pandas+SQLAlchemy	中小型数据（<10GB）	Python	灵活易调试，但内存占用高
PyMySQL+多线程分片	高并发大表	Python	速度快，需处理线程安全
pg_dump/pg_restore	PostgreSQL => PostgreSQL	Shell脚本	原生支持，但无法执行复杂转换
DataX（阿里巴巴）	异构数据库	Java+Python	稳定高效，但学习成本较高
推荐组合	通用场景	Python+PyMySQL+SQLAlchemy	兼顾性能与可维护性

核心原则：选择脚本语言时，优先考虑团队熟悉度，对于大数据量（>100GB），建议使用DataX或Apache Sqoop；对于中小规模，Python脚本足以满足需求。

从零搭建批量迁移脚本（核心代码示例）

假设我们需要将MySQL数据库中的orders表（1000万条数据）迁移到PostgreSQL，以下脚本实现了全量迁移+增量同步的核心逻辑：

# -*- coding: utf-8 -*-
import pymysql
import psycopg2
from datetime import datetime
import threading
from queue import Queue
# 数据库连接配置
mysql_config = {
    'host': '192.168.1.100',
    'port': 3306,
    'user': 'root',
    'password': 'password123',
    'database': 'source_db',
    'charset': 'utf8mb4'
}
pg_config = {
    'host': '192.168.1.200',
    'port': 5432,
    'user': 'admin',
    'password': 'pg_pass',
    'database': 'target_db'
}
# 批量迁移函数（分片读取）
def batch_migrate(table_name, chunk_size=10000):
    # 源库连接
    src_conn = pymysql.connect(**mysql_config)
    # 目标库连接
    dst_conn = psycopg2.connect(**pg_config)
    try:
        # 获取总行数
        with src_conn.cursor() as cursor:
            cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
            total_rows = cursor.fetchone()[0]
        # 分批读取并写入
        offset = 0
        while offset < total_rows:
            with src_conn.cursor(pymysql.cursors.DictCursor) as cursor:
                cursor.execute(f"SELECT * FROM {table_name} LIMIT {chunk_size} OFFSET {offset}")
                rows = cursor.fetchall()
            # 批量插入pg
            with dst_conn.cursor() as pg_cursor:
                pg_cursor.executemany(
                    f"INSERT INTO {table_name} (id, order_date, amount) VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
                    [(row['id'], row['order_date'], row['amount']) for row in rows]
                )
            dst_conn.commit()
            offset += chunk_size
            print(f"已迁移 {min(offset, total_rows)} / {total_rows} 条")
    finally:
        src_conn.close()
        dst_conn.close()
# 多线程加速迁移（适用于大表）
def concurrent_migrate(table_name, thread_num=4):
    # 获取表的id范围
    src_conn = pymysql.connect(**mysql_config)
    with src_conn.cursor() as cursor:
        cursor.execute(f"SELECT MIN(id), MAX(id) FROM {table_name}")
        min_id, max_id = cursor.fetchone()
    src_conn.close()
    # 分片计算：每个线程负责的id段
    step = (max_id - min_id) // thread_num
    ranges = [(min_id + i*step, min_id + (i+1)*step - 1) for i in range(thread_num)]
    ranges[-1] = (ranges[-1][0], max_id)
    def worker(id_range):
        # 每个线程独立处理自己的id段
        src_conn = pymysql.connect(**mysql_config)
        dst_conn = psycopg2.connect(**pg_config)
        try:
            with src_conn.cursor(pymysql.cursors.DictCursor) as cursor:
                cursor.execute(f"SELECT * FROM {table_name} WHERE id BETWEEN {id_range[0]} AND {id_range[1]}")
                rows = cursor.fetchall()
            with dst_conn.cursor() as pg_cursor:
                pg_cursor.executemany(
                    f"INSERT INTO {table_name} VALUES (%(id)s, %(order_date)s, %(amount)s)",
                    rows
                )
            dst_conn.commit()
        finally:
            src_conn.close()
            dst_conn.close()
    threads = [threading.Thread(target=worker, args=(r,)) for r in ranges]
    for t in threads: t.start()
    for t in threads: t.join()
    print("多线程迁移完成")

说明：