Python脚本化数据库备份、导出与迁移的完整实践

做系统运维和数据开发的朋友，应该都遇到过这种尴尬：半夜收到磁盘告警，登上去一看，备份文件把空间塞满了；或者业务方要一份上个月的订单明细，你下意识写了一条select * from orders扔给 pandas，结果生产库的内存直接被拉爆；再或者公司要做数据库平台迁移，单表几个亿的数据，迁移工具跑了两天，最后对账时发现行数对不上，又得从头来。数据库的备份、导出、迁移这三件事，说起来是基本功，但真正在生产环境里做好，每一个环节都可能踩坑。我的做法是用Python把这三类操作全部脚本化，让机器按照既定流程执行，我只需要看日志和结果。这篇文章就把我实际用过的方案、代码和踩过的坑整理出来，给正在做同样事情的人一个可以抄作业的参考。

不论你是运维、后端开发，还是数据工程师，只要你的工作里要常和数据库打交道，这篇文章都值得读完。我会先讲清楚三件事的区别，再分别给出备份、大文件导出、跨库迁移的完整脚本思路，最后补充工程化和排错的经验，尽量做到看完了就能在自己的环境里跑起来。

1. 先把概念理清楚：备份、导出、迁移是三件不同的事

1.1 备份、导出、迁移的核心差异

很多人会把“备份”和“导出”混为一谈，也会把“迁移”当成“导出加导入”。表面上看，都是把数据从库里读出来再写到别的地方，但它们在目标、产物和验证方式上有本质区别。

备份关心的核心是“恢复点目标RPO”和“恢复时间目标RTO”，所以备份文件必须冗余保存、定期验证恢复。导出关心的核心是“数据格式要能被下游消费”，比如给业务方的Excel报表，导出的CSV字段分隔符不对，人家打开就是乱的。迁移关心的核心则是“目标库的表结构、数据、约束和应用行为必须和源库保持一致”，字段类型不兼容、字符集不同、外键顺序颠倒，都会在切换后引发线上故障。

理解了这些差异，你就能理解为什么我不建议把三个功能写成一个“万能脚本”。三个功能混在一起，出问题时很难快速定位，备份失败可能会影响迁移流程，导出的临时文件也可能被误当成备份删除。我的习惯是拆成三个独立模块，共用一套配置文件，再通过调度脚本串联。

1.2 自动化脚本开始前，先把这些参数定好

在写第一行代码之前，我建议你先花半天时间把下面这些信息梳理清楚，否则脚本写一半很容易返工。

• 数据库连接信息：host、port、用户名、密码、要操作的库名。密码一定不要写死在代码里，建议使用环境变量或配置管理工具。
• 备份目录和磁盘空间：备份文件放哪里，所在分区剩余空间有多少，因为备份前后文件大小可能差距很大。
• 备份保留策略：本地保留几天？是否要异地备份？是全量备份还是全量加增量？
• 执行窗口：备份和迁移都尽量选业务低峰期，避免锁表或IO争用影响线上。
• 目标库信息：如果是迁移，目标库类型、版本、字符集、编码规则，以及是否允许停机切换。
• 网络和权限：迁移和导出通常要跨服务器访问，网络带宽瓶颈在哪，数据库账号是否具备对应的只读或写权限。

把这些信息整理成一份环境清单，脚本只是把这份清单落地执行的工具。没有这份清单，脚本做得再漂亮，到生产环境也会因为权限、路径、字符集各种问题跑不起来。

2. 数据库备份自动化：全量备份与增量备份的取舍

2.1 用Python调度mysqldump实现全量逻辑备份

数据库备份里最常见的还是MySQL，而提到MySQL备份，很多人第一反应就是mysqldump。虽然有物理备份工具如Percona XtraBackup，但mysqldump这种逻辑备份在跨版本恢复、整库迁移时还是最通用的方案。

逻辑备份和物理备份的取舍是这样的：

我自己的经验是：几个GB以内的库，mysqldump完全够用；几十GB以上就要考虑物理备份或者基于binlog的增量方案。下面的Python脚本是我常用的全量备份模板，用subprocess调用mysqldump，边导边用gzip压缩，然后按日期生成文件名。

import os import gzip import subprocess from pathlib import Path from datetime import datetime def backup_mysql(host, port, user, password, dbname, backup_dir, keep_days=7): backup_path = Path(backup_dir) backup_path.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = backup_path / f"{dbname}_{timestamp}.sql.gz" env = os.environ.copy() # 使用MYSQL_PWD避免密码出现在ps输出中 env["MYSQL_PWD"] = password cmd = [ "mysqldump", f"--host={host}", f"--port={port}", f"--user={user}", "--single-transaction", "--quick", "--routines", "--triggers", dbname, ] try: with gzip.open(filename, "wb") as fout: subprocess.run(cmd, stdout=fout, env=env, check=True) print(f"[OK] 备份完成: {filename}") except subprocess.CalledProcessError as e: print(f"[ERROR] mysqldump执行失败: {e}") return False # 清理过期备份 cleanup_old_backups(backup_path, keep_days, dbname) return True def cleanup_old_backups(backup_path, keep_days, dbname): cutoff = datetime.now().timestamp() - keep_days * 24 * 3600 for f in backup_path.glob(f"{dbname}_*.sql.gz"): if f.stat().st_mtime < cutoff: f.unlink() print(f"[CLEAN] 删除旧备份: {f}")

这里有几个参数值得说明。

--single-transaction是InnoDB引擎下保证备份一致的利器，它让mysqldump基于一个事务快照备份，不加锁，不阻塞线上读写。如果你的表是MyISAM引擎，这个参数不生效，备份时会锁表，所以生产环境我强烈建议都用InnoDB。

--quick让mysqldump逐行从服务器拉取数据而不是先缓存到内存，对大数据量表更友好。--routines和--triggers是很多人容易漏的，不加上它们，存储过程和触发器不会被备份，恢复后应用直接报错。

用subprocess而不是直接用os.system，可以避免shell命令拼接带来的注入和转义问题，尤其当数据库名或文件名里有空格、特殊字符时，参数列表方式更安全。

2.2 备份保留策略与定时调度

有了备份脚本还不够，你得让它按时跑起来，并且自动清理旧文件。上面代码里已经留了cleanup_old_backups函数，按文件修改时间删除保留天数之前的备份。这里有一个建议：保留周期不要拍脑袋定，要根据业务对数据丢失的容忍度和存储成本综合决定。

如果每天凌晨做一次全量备份，保留最近7天，磁盘占用就是一周全量备份的总和。如果还要做月度归档，那把月度备份文件移动到另一个目录或对象存储，不要和日常备份混在一起。

定时调度在Linux下用crontab，在Windows下用计划任务程序。下面是一个crontab示例：

0 2 * * * cd /opt/db_automation && /usr/bin/python3 backup_mysql.py >> logs/backup.log 2>&1

这条计划表示每天凌晨两点执行备份脚本，日志写进backup.log。日常运行中我发现，一定要在调度命令里加上cd切换到脚本所在目录，否则脚本里用的相对路径很容易出问题。日志文件也要按日期滚动，不然几个月后单个日志文件会非常大。

2.3 备份验证：从“做了备份”到“真的能恢复”

很多人备份完看一眼文件大小没问题就不管了，直到某一天数据库真的挂了，才发现备份文件早就损坏或者内容不完整。我在这件事上栽过跟头，所以现在我的脚本里，备份完成后至少会做两层验证。

第一层是快速健康检查：备份文件不是0字节，文件里的gzip压缩流完整，能解压且包含建表语句。这个检查可以放在备份脚本里自动执行，比如用Python的zipfile/gzip模块打开文件，同时用grep搜索文件里是否存在关键表的CREATE TABLE关键字。

第二层是恢复演练：把最新的备份文件导入到一个临时数据库，对比几张核心表的行数是否和源库一致。这个步骤不一定要每天做，但至少每周做一次。脚本里可以调用mysql命令导入备份到临时库，然后执行计数查询。

import gzip import subprocess def quick_check(backup_file, expected_table): with gzip.open(backup_file, "rb") as f: content = f.read(1024 * 1024) if expected_table.encode() not in content: raise RuntimeError("备份文件中未找到预期表结构") return True

快速检查只能发现明显问题，真正的保障还是定期的恢复演练。这个过程很费时间，但关键时刻能救命。我给自己的要求是：每月至少完整恢复演练一次，并且记录恢复耗时，作为RTO评估的数据。

3. Python导出大文件：别让内存先爆炸

3.1 一次读全表，内存为什么扛不住

导出数据，尤其是大表导出，最常见的坑就是一次性把查询结果全部load到内存里。

很多人一开始会这么写：

import pandas as pd import pymysql conn = pymysql.connect(host="localhost", user="root", password="...", database="mydb") df = pd.read_sql("SELECT * FROM big_table", conn) df.to_csv("big_table.csv", index=False)

这段代码看着简单，但假如big_table有3000万行、每行平均200字节，查询结果到内存中会占6GB以上，再加上pandas内部的索引和列转换，实际占用经常翻倍，16GB内存的机器都会被瞬间打爆。

根源在于数据库客户端默认是一次性把整个结果集拉到客户端。无论你用什么语言什么库，只要没有开启流式游标，内存就会被大查询拖垮。解决思路是让数据库服务器分批发送数据，客户端处理完一批再取下一批。

3.2 使用服务端游标流式导出CSV

在Python中，pymysql提供了SSCursor，也就是服务端游标。使用它执行SELECT时，客户端不会立刻拉取全部数据，而是按需从服务器获取。配合fetchmany分批处理，可以稳定导出远大于内存的数据量。

下面是我常用的流式导出CSV示例：

import csv import pymysql from pymysql.cursors import SSCursor def export_large_table(host, port, user, password, database, table, output_file): conn = pymysql.connect( host=host, port=port, user=user, password=password, database=database, cursorclass=SSCursor ) cursor = conn.cursor() sql = f"SELECT * FROM {table}" cursor.execute(sql) with open(output_file, "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) # 写入表头 writer.writerow([desc[0] for desc in cursor.description]) while True: rows = cursor.fetchmany(1000) if not rows: break writer.writerows(rows) cursor.close() conn.close() print(f"[OK] 导出完成: {output_file}")

这里有两个细节需要注意。

第一，cursorclass=SSCursor必须在建立连接时指定。如果连接没有指定，后面单纯cursor = conn.cursor(pymysql.cursors.SSCursor)也行，但前者更清晰。第二，导出文件编码我推荐用utf-8-sig，这样用Excel直接打开CSV时中文不会乱码，但如果下游是Linux数据仓库，建议改成普通utf-8，避免BOM头干扰。

如果你用的是PostgreSQL，SQLAlchemy里同样可以开启流式查询：

from sqlalchemy import create_engine import csv engine = create_engine("postgresql://user:pass@host:5432/db", execution_options={"stream_results": True}) conn = engine.connect() result = conn.exec_driver_sql("SELECT * FROM big_table").execution_options(yield_per=1000) with open("big_table.csv", "w", newline="") as f: writer = csv.writer(f) writer.writerow(result.keys()) for row in result: writer.writerow(row) conn.close()

yield_per告诉SQLAlchemy每1000行批量获取一次，避免全部结果驻留内存。

3.3 导出格式选择和校验

CSV能解决90%的大文件导出需求，但并不是所有场景都适合用CSV。

尤其注意Excel这个坑。E某cel最多年限是104万行左右，超过这个行数你就算用pandas.to_excel能打开，也会被截断，而且生成的xlsx文件往往巨大。我见过有人花了三个小时导出一张200万行的Excel，最后业务方说“怎么只有一半”，这就是对格式边界不熟悉导致的。针对大表导出，还是老老实实走CSV或Parquet。

导出完成后，还有一个经常被忽略的步骤：校验。最简单的校验是统计导出文件的行数，和数据库里count(*)的结果做对比。对于CSV文件，可以这样数：

def count_lines(filepath): with open(filepath, "r", encoding="utf-8") as f: return sum(1 for line in f)

注意CSV里有换行字段时，这种数行方法不准确，需要用csv.reader逐行读取计数。更严谨的做法是在导出时就计算每行的checksum，生成一个汇总值，迁移完成后在目标库再算一次汇总值，两边一致才能说数据完整。

3.4 顺带一提：表结构和ER图自动化

很多人用数据库工具导出ER图，其实这事也可以用Python自动化。用sqlacodegen可以从现有数据库反向生成SQLAlchemy模型，再用graphviz之类的库生成ER图。虽然这不算备份导出迁移的核心功能，但做文档自动化时很实用。

pip install sqlacodegen sqlacodegen mysql+pymysql://user:pass@host/dbname --outfile models.py

生成的models.py里定义了所有表结构，再结合er_alchemy之类的库，输入模型文件即可输出可视化的ER图。这个方案的好处是每次表结构变更后快速重新生成文档，避免手工画图跟不上线上变化。

4. 数据库迁移自动化：从同构到异构

4.1 迁移前必须做的四件事：盘点、映射、试迁移、切换计划

数据库迁移是我觉得最有挑战的部分，因为失败后的影响是直接面向业务用户的。我的经验是，迁移前必须先做下面四件事，不做完不要碰代码。

第一，数据盘点。统计源库里有多少库、每个库多少表、总数据量多大、哪些表是核心的、哪些是日志表可以清理。这一层可以通过查询information_schema.tables得到。

第二，字段类型映射。不同数据库的字段类型不是一一对应的。比如MySQL的tinyint在PostgreSQL里对应smallint，MySQL的varchar(n)在ClickHouse里要转成String或FixedString。提前列一张映射表，迁移脚本里统一做类型转换，比在SQL里反复改要靠谱得多。

第三，试迁移。用小批量数据先跑一遍完整流程，记录耗时、确认约束和外键。试迁移不是做样子，它要测出真实迁移速率的基线，这样你才能估算全量迁移需要多少时间，是否能在维护窗口内完成。

第四，切换计划。明确的停机开始时间、结束时间、回滚方案和负责人。数据库迁移最怕的是切完发现应用不兼容，又找不到回滚方案。脚本只能处理数据，人要为决策负责。

4.2 同构库迁移的Python脚本

如果源库和目标库都是MySQL，迁移相对简单，可以用下面的脚本批量搬表。

import pymysql def migrate_mysql_table(src_conn, dst_conn, table_name, batch_size=500): cur_src = src_conn.cursor(pymysql.cursors.SSCursor) cur_src.execute(f"SELECT * FROM `{table_name}`") cols = [desc[0] for desc in cur_src.description] col_names = ",".join(cols) placeholders = ",".join(["%s"] * len(cols)) insert_sql = f"INSERT INTO `{table_name}` ({col_names}) VALUES ({placeholders})" total = 0 while True: rows = cur_src.fetchmany(batch_size) if not rows: break with dst_conn.cursor() as cur_dst: cur_dst.executemany(insert_sql, rows) dst_conn.commit() total += len(rows) print(f"[{table_name}] 已迁移 {total} 行") cur_src.close() return total

这里批量大小batch_size建议取500到1000之间。太小时事务数量太多，太大会导致单条SQL过长，数据库解析压力大。如果目标库也要保留自增id，插入语句里要把id字段一起带上，并在目标库先关闭自增约束，或者用SET sql_log_bin=0这一类配置减少日志开销。

同构迁移看似简单，但索引和外键顺序很关键。源表如果有自增id、唯一索引和外键，目标表最好先按源表结构创建好，再关闭外键检查导入数据，最后打开。否则插入顺序稍有不慎，外键校验就会让导入卡死。

4.3 异构迁移：MySQL到ClickHouse整体迁移示例

最近碰到不少把MySQL数据整体迁移到ClickHouse做分析加速的场景。ClickHouse是列式存储，迁移时不能用普通insert一条条搬，那样速度太慢。正确方式是把MySQL表导出为CSV，再用ClickHouse的FORMAT CSV批量导入。

配合Python流式导出到CSV，然后调用clickhouse-client导入，是我实测过最稳定的路线。

clickhouse-client \ --host clickhouse_host \ --query "INSERT INTO mydb.mytable FORMAT CSV" \ --input_format_with_names_use_header=1 \ < mytable.csv

用Python驱动也可以，clickhouse_driver库提供了insert方法，传入由行组成的列表，效率不错。但无论用哪种方式，有几类坑必须提前处理。

一条是字段类型映射。MySQL的timestamp在ClickHouse里通常映射为DateTime，但可能需要处理时区；MySQL的NULL空字符串，ClickHouse里要决定是存NULL还是空字符串，这两者在分析时语义不同。第二条是特殊字符。CSV导出时必须处理好字段中换行、逗号、引号，否则ClickHouse导入会错位。Python的csv模块默认就能处理转义，但如果你手工拼字符串导出，很容易踩坑。第三条是排序和压缩。ClickHouse表可以指定ORDER BY和ENGINE，如果目标表设计得不对，导入后查询性能可能比MySQL还差。迁移前先设计好MergeTree表结构，再做数据搬运。

4.4 Python环境和二进制依赖迁移经验

数据库迁移不光是数据本身的迁移，运行里的Python脚本、驱动、虚拟环境同样有坑。刚入行的时候我经常遇到“同一个脚本在开发机上好好的，到生产机器上ModuleNotFoundError”，大多数原因是虚拟环境没有真正迁移过去。

Python虚拟环境迁移，推荐两条经验。

如果源环境是conda管理的，用conda env export导出完整环境描述，目标机器用conda env create -f environment.yml重建。

conda env export --name myenv > environment.yml

如果用的是venv或纯pip环境，用pip freeze > requirements.txt导出依赖列表，目标机器重建后通过pip install -r requirements.txt安装。但要注意，pip freeze会带上本地平台相关的包名，如果源机器是x86而目标是ARM，某些带C扩展的包版本可能装不上或行为不一致。

碰到.so或.pyd这类编译型扩展，比如pymysql的加密模块、或者某种原生分析库，基本没有靠拷贝文件迁移的可能，必须要在目标平台上用对应版本的Python重新编译安装。我自己在x86到ARM的迁移中踩过一次坑，当时图省事直接把整个site-packages目录拷过去，结果所有需要C编译的包全部导入失败。后来老老实实重建虚拟环境，问题立刻消失。

5. 自动化脚本的工程化落地与排错

5.1 配置文件与敏感信息管理

备份、导出、迁移脚本如果要在多个环境切换，连接信息最好放在配置文件里，不要写死在脚本里。我常使用的格式是YAML，简单直观。

mysql: host: 127.0.0.1 port: 3306 user: backup_user password: "${MYSQL_PASSWORD}" database: mydb backup: dir: /data/backup keep_days: 7 migration: source: mysql target: clickhouse batch_size: 1000

密码通过环境变量引用，比如上面用${MYSQL_PASSWORD}，在Python里用os.path.expandvars展开。这样配置文件即使不小心提交到git，也不会直接暴露秘密。更严格的做法是使用Vault之类的密钥管理工具，但对大部分团队来说，环境变量已经足够。

5.2 日志、异常通知与重试

自动化脚本最怕的就是半夜静默失败。你早上醒来才发现昨天的备份根本没生成，那问题就大了。所以每次执行都要有日志，异常一定要通知。

Python的logging模块足够用：

import logging logging.basicConfig( filename="logs/backup.log", level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__)

在一些关键节点，比如备份完成、文件清理完成、迁移批量提交完成，都记录一条INFO日志。捕获到异常时，除了记录error日志，还可以发送告警到企业微信群或邮件。最简单的方案是用requests调用webhook，几行代码就能搞定：

def send_alert(message): requests.post("https://your-webhook-url", json={"content": message})

重试机制要针对具体情况设计。网络抖动导致的连接失败可以重试两到三次，但数据导入中途失败就不能盲目重试，否则可能出现重复数据。我的习惯是：幂等操作可以重试，非幂等操作失败只能报警人工介入。

5.3 常见问题速查表

下面是我在备份、导出、迁移中遇到过的典型问题，整理成速查表。

针对Windows批处理那一条，我想多说一句。很多人在bat文件里写mysqldump ... > D:\backup\file.sql，但当D盘不存在或路径写成了D:\backup\后面漏了目录，系统就会报“The system cannot write to the specified device”。解决办法是确认目录存在、路径里不要带引导和被保留字符，最好在bat里先判断文件目录是否存在再执行重定向。如果你用的是Python脚本，尽量用open/write而不是在bat里重定向，这样报错信息会准确得多。

5.4 一些更偏门但重要的经验

除了上面这些，还有几个容易被忽视的小经验，值得聊聊。

一是不要在代码里用字符串拼接shell命令。哪怕你只是调用subprocess.call("mysqldump -u" + user + " -p" + password + " " + dbname, shell=True)，遇到数据库名带空格或特殊字符时轻则命令报错，重则引发安全问题。始终用参数列表形式，让subprocess帮你处理转义。

二是备份前一定要确认磁盘空间。脚本里可以先计算备份目录当前可用空间，再估算备份文件可能的大小。虽然估算不一定准，但如果临时分区小于数据库体积的1.5倍，我会直接告警，坚决不做无准备的备份。

三是迁移之前做一次小流量试迁移，不仅验数据，还要验速度。有一次我迁移一个8亿行的表，原本预计4小时能完成，试迁移后才发现源库是共享IO，大批量读取会拖垮线上，最后只能把迁移时间拉长到半夜单独跑。没有试迁移，这个问题在正式切换当天才会暴露，那场面就很被动了。

四是迁移完成后要做数据一致性校验，但不要只数行数。行数一致不代表数据一致，字段值的分布可能早就错位。我的办法是选择几个关键业务表，对几个数值列做SUM或CHECKSUM，再对比源库和目标库的结果。如果关键数值能对上，整体可信度就高很多。

最后再说几句体己话

数据库自动化做到后面，拼的不是脚本写得多炫，而是流程设计得稳。备份、导出、迁移每件事拆开做，每完成一步都有日志、有校验、有告警，这样系统才能脱离人工盯守长期运转。我现在的习惯是：无论多简单的操作，都要先写一个--dry-run参数，只打印将要执行的动作和预期影响，确认无误后再真正执行。这个习惯帮我避免了很多手误，也让我能在紧急时刻冷静核对脚本行为。等你把备份、导出、迁移这些常规操作全部自动化之后，你会发现运维工作真正的重点变成了对流程的持续优化，而不是每天重复操作数据库客户端。