如何编写自动迁移数据脚本

wen 实用脚本 2026-07-25 1

从零构建高效、可靠的数据迁移系统

目录导读

数据迁移的痛点与核心挑战
自动迁移脚本的架构设计原则
完整脚本编写流程与代码片段示范
常见错误规避与性能优化技巧
问答环节：从理论到实战的深度解析

数据迁移的痛点与核心挑战

在企业IT运维中,数据迁移是一项高频且高风险的任务，无论是从旧系统迁移至新平台、数据库版本升级，还是跨云数据同步，手动操作极易引发数据丢失、格式错乱或服务中断，根据Stack Overflow 2024年开发者调查，超过62%的数据库管理员表示曾因手动迁移脚本导致至少一次生产事故。

如何编写自动迁移数据脚本

核心挑战包括：

数据一致性：迁移过程中如何确保源端与目标端数据完全一致？
断点续传：当网络中断或任务失败时，如何从失败点恢复而非从头开始？
性能瓶颈：大量数据写入时如何避免锁表或内存溢出？
类型映射：不同数据库间的字段类型（如MySQL的DATETIME与PostgreSQL的TIMESTAMP）如何自动转换？

解决方案：编写一个具备监控、重试、校验和日志记录的自动迁移脚本，而非一次性SQL脚本。

自动迁移脚本的架构设计原则

高质量的迁移脚本应遵循以下五大设计原则：

幂等性：同一脚本执行多次应保持最终结果一致（例如使用INSERT ... ON DUPLICATE KEY UPDATE或先删除再插入策略）。
分片与批处理：将数据分割成固定大小的块（如每批次1000条），避免全表锁定。
日志与审计：记录每批次的开始时间、处理行数、异常详情，便于事后排查。
可配置化：通过外部配置文件（如YAML、JSON）定义源库、目标库、表映射关系，而非硬编码。
监控与告警：内置成功/失败计数、耗时统计，并支持邮件或Webhook通知。

示例架构图（文字版）：

[源数据库] → [读取模块] → [数据转换器] → [写入模块] → [目标数据库]
                                    ↑
                             [错误队列] → [重试器] → [死信处理]

完整脚本编写流程与代码片段示范

以Python为例,演示一个从MySQL迁移至PostgreSQL的自动脚本核心设计。

1 环境准备与依赖

import psycopg2     # PostgreSQL驱动
import pymysql      # MySQL驱动
from datetime import datetime
import logging
import json

2 核心迁移逻辑（分批次+断点续传）

class DataMigrator:
    def __init__(self, config_file='migrate_config.json'):
        with open(config_file) as f:
            self.cfg = json.load(f)
        self.source_conn = pymysql.connect(**self.cfg['source'])
        self.target_conn = psycopg2.connect(**self.cfg['target'])
        self.batch_size = self.cfg.get('batch_size', 1000)
        self.resume_point = self.load_checkpoint()  # 从文件读取上次位置
    def transfer_table(self, table_name, columns):
        cursor = self.source_conn.cursor()
        offset = self.resume_point.get(table_name, 0)
        while True:
            query = f"SELECT {','.join(columns)} FROM {table_name} LIMIT {self.batch_size} OFFSET {offset}"
            cursor.execute(query)
            rows = cursor.fetchall()
            if not rows:
                break  # 数据全部迁移完成
            # 批量写入目标库（使用executemany或COPY命令）
            self.batch_insert(table_name, columns, rows)
            offset += len(rows)
            self.save_checkpoint(table_name, offset)
            logging.info(f"Table {table_name}: processed {offset} rows")
    def batch_insert(self, table, cols, data):
        # 使用PostgreSQL的COPY命令提高写入速度
        import io
        buffer = io.StringIO()
        for row in data:
            buffer.write('\t'.join([str(v) if v else 'NULL' for v in row]) + '\n')
        buffer.seek(0)
        cursor = self.target_conn.cursor()
        cursor.copy_from(buffer, table, sep='\t', null='NULL', columns=cols)
        self.target_conn.commit()

3 关键优化点

使用COPY而非INSERT：PostgreSQL的COPY命令写入速度是普通INSERT的50倍以上。
批量读取而非游标：对于MySQL，使用LIMIT...OFFSET（需确保有索引，避免全表扫描减速）。
异步写入错误队列：若某批次失败，记录至独立文件，继续处理后续批次。

常见错误规避与性能优化技巧

1 错误场景与解决方案

错误类型	现象	解决方案
主键冲突	重复数据导致脚本中断	使用`ON CONFLICT DO UPDATE`或`INSERT IGNORE`
网络超时	长连接断开	设置连接池，并添加重试机制（指数退避算法）
字符集不兼容	中文字符乱码	统一使用UTF-8，并在连接参数中指定字符集
内存溢出	一次性加载百万行数据	强制分批读取，每批次处理完成后释放内存

2 性能调优实战

索引策略：源表需为ORDER BY或OFFSET字段建立索引，否则LIMIT...OFFSET会越来越慢。
并行管道：对多张无依赖关系的表，使用concurrent.futures.ThreadPoolExecutor并行迁移。
数据压缩：迁移前对文本字段进行GZip压缩，传输后解压，可降低70%网络消耗。

问答环节：从理论到实战的深度解析

Q1：迁移过程中如果目标库突然宕机，脚本如何保证数据不丢失？
A：在每批次写入目标库前，先将该批次数据写入本地临时文件（如batch_123.tmp），写入目标成功后删除文件，若宕机重启，脚本优先检测临时文件并恢复未确认的批次，所有操作在事务中进行，确保原子性。

Q2：如何处理源库与目标库字段类型不一致的问题？
A：在配置文件中增加类型映射字典。

TYPE_MAP = {
    'MySQL.DATETIME': 'PostgreSQL.TIMESTAMP',
    'MySQL.TINYINT': 'PostgreSQL.BOOLEAN',
    'MySQL.INT': 'PostgreSQL.BIGINT'
}

在读取阶段自动转换数据类型,并在写入前统一格式化。

Q3：迁移大表（如10亿行）时，如何监控实时进度？
A：使用多线程记录器，每秒输出当前处理行数、剩余行数、每秒传输速率，公式：剩余时间 = (总行数 - 已完成行数) / 平均速率，同时通过Webhook将进度推送到监控面板（如Prometheus + Grafana）。

Q4：脚本完成后如何进行数据一致性校验？
A：迁移后执行哈希校验：

计算源表所有行的MD5聚合值：SELECT MD5(GROUP_CONCAT(CONCAT_WS('|', col1, col2) ORDER BY id)) FROM table
在目标表执行同样的计算,若结果一致，则验证通过，为提高性能，可对每1000行计算局部哈希后合并。

Q5：有没有推荐的现成工具，还是必须从零编写？
A：若源库和目标库异构，且需要高度定制化逻辑（如字段加密、丢弃脏数据），建议编写脚本，若结构相同，可优先使用开源工具：

pgloader（MySQL→PostgreSQL 行业标准）
AWS DMS（跨云数据库迁移服务）
Apache Sqoop（Hadoop与传统数据库间批量传输）

本文提供的脚本框架可集成这些工具作为底层引擎,实现更复杂的编排。

注意：上述代码中的域名（如example.com）已按规范替换为示例名称，实际使用时请替换为真实环境配置。