Navicat右键导出2.3GB数据库跑47分钟断了三次换mydumper四线程6分钟搞定，数据库导出速度差近8倍怎么选

数据库导出这件事有一个很容易被忽略的规律：导出速度的差距不是线性的，是指数级的。几百兆的小库，用Navicat右键导出和用命令行mysqldump，时间差不多——都是十几秒。但一旦数据库超过1GB，命令行工具和图形化工具的差距立刻拉开：图形化工具需要先把数据加载到内存、再渲染到界面、再写入文件，每一步都是瓶颈；命令行工具直接走数据库底层协议流式写入磁盘，中间没有内存中转。而到了10GB以上的大库，连mysqldump都不够用了，必须上mydumper或SELECT INTO OUTFILE这种并行或文件系统级别的方案。

选错工具不光是慢的问题——大表导出过程中内存溢出、连接超时、锁表导致业务中断、导完发现字符编码全乱了，这些坑每一个都能让你多花半天时间。而且不同数据库类型（MySQL、PostgreSQL、MongoDB、SQLite）的导出工具和命令完全不同，没有一个"万能导出工具"能搞定所有场景。

mysqldump：最通用的方案，但大库扛不住

mysqldump是MySQL自带的命令行导出工具，所有MySQL环境都能用，不需要额外安装。单表几百万条数据以内表现稳定，但超过千万行后单线程逐表导出的短板就暴露了——一张大表导出可能需要几十分钟，而且导出的SQL文件导入回去也要同等甚至更长时间。

三个必加参数：--single-transaction（InnoDB不锁表，保证一致性读）、--quick（逐行读取而非全部加载到内存）、--default-character-set=utf8mb4（防止中文乱码）。少了任何一个，轻则导出失败，重则锁表导致线上业务中断。

mydumper：并行导出，大库场景速度翻几倍

mydumper是mysqldump的多线程替代品，C++编写，底层机制完全不同：它不是逐表导出，而是把每张表按主键分成多个数据块（chunk），然后多个线程同时导出不同的块。一张2GB的大表，mydumper开4个线程，每个线程导出500MB的数据块，理论上时间是mysqldump的四分之一。

关键参数：-t 4（4个线程）、-r 1000000（每100万行一个文件）、--rows-per-chunk 100000（每个chunk包含10万行）。线程数不是越多越好，建议设为CPU核心数的一半到等于核心数之间。

SELECT INTO OUTFILE：最快但最受限

这条SQL语句让MySQL直接把查询结果写成文件放在服务器磁盘上，不经过客户端网络传输，速度是所有方案里最快的。但它有三个硬伤：需要FILE权限（很多云数据库禁用了）、文件生成在服务器端（你还需要再下载）、格式是纯CSV（不含表结构）。适合做数据分析导出，不适合做备份迁移。

MySQL批量导出Shell脚本（三种模式一键切换）

#!/bin/bash# MySQL批量导出脚本 - 支持mysqldump/mydumper/SELECT INTO OUTFILE三种模式DB_USER="root"DB_PASS="your_password"DB_HOST="localhost"OUTPUT_DIR="./db_export_$(date +%Y%m%d_%H%M%S)"MODE="${1:-mysqldump}"  # 默认mysqldump，可选 mydumper / outfileTHREADS=4mkdir -p "$OUTPUT_DIR"# ============ 模式1：mysqldump批量导出（全库所有表） ============dump_all_tables_mysqldump() {echo "=== 模式：mysqldump 全库导出 ==="# 获取所有数据库列表（排除系统库）databases=$(mysql -u"$DB_USER" -p"$DB_PASS" -h"$DB_HOST" -N -e \"SELECT SCHEMA_NAME FROM information_schema.SCHEMATAWHERE SCHEMA_NAME NOT IN ('information_schema','mysql','performance_schema','sys')")for db in $databases; doecho "  导出数据库: $db"mysqldump -u"$DB_USER" -p"$DB_PASS" -h"$DB_HOST" \--single-transaction \--quick \--default-character-set=utf8mb4 \--routines --triggers --events \"$db" > "$OUTPUT_DIR/${db}.sql"if [ $? -eq 0 ]; thensize=$(du -h "$OUTPUT_DIR/${db}.sql" | cut -f1)echo "    ✓ ${db}.sql ($size)"elseecho "    ✗ ${db} 导出失败"fidone}# ============ 模式2：mydumper并行导出（大库推荐） ============dump_all_tables_mydumper() {echo "=== 模式：mydumper 并行导出（${THREADS}线程） ==="databases=$(mysql -u"$DB_USER" -p"$DB_PASS" -h"$DB_HOST" -N -e \"SELECT SCHEMA_NAME FROM information_schema.SCHEMATAWHERE SCHEMA_NAME NOT IN ('information_schema','mysql','performance_schema','sys')")for db in $databases; doecho "  导出数据库: $db"mydumper \-u "$DB_USER" -p "$DB_PASS" -h "$DB_HOST" \-B "$db" \-t "$THREADS" \-r 1000000 \--rows-per-chunk 100000 \-o "$OUTPUT_DIR/${db}_mydumper" \--compressif [ $? -eq 0 ]; thenecho "    ✓ ${db} 导出完成（mydumper格式）"elseecho "    ✗ ${db} 导出失败"fidone}# ============ 模式3：SELECT INTO OUTFILE（数据分析用） ============dump_all_tables_outfile() {echo "=== 模式：SELECT INTO OUTFILE（CSV格式，最快） ==="databases=$(mysql -u"$DB_USER" -p"$DB_PASS" -h"$DB_HOST" -N -e \"SELECT SCHEMA_NAME FROM information_schema.SCHEMATAWHERE SCHEMA_NAME NOT IN ('information_schema','mysql','performance_schema','sys')")for db in $databases; dotables=$(mysql -u"$DB_USER" -p"$DB_PASS" -h"$DB_HOST" -N -e \"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA='$db'")for table in $tables; dooutfile="/var/lib/mysql-files/${db}_${table}.csv"mysql -u"$DB_USER" -p"$DB_PASS" -h"$DB_HOST" "$db" -e \"SELECT * INTO OUTFILE '$outfile'FIELDS TERMINATED BY ','OPTIONALLY ENCLOSED BY '\"'LINES TERMINATED BY '\n'FROM \`$table\`;"if [ $? -eq 0 ]; thenecho "    ✓ ${db}.${table} → $outfile"fidonedone}# 执行case "$MODE" inmysqldump) dump_all_tables_mysqldump ;;mydumper)  dump_all_tables_mydumper ;;outfile)   dump_all_tables_outfile ;;*)         echo "用法: $0 {mysqldump|mydumper|outfile}" ;;esacecho ""echo "导出完成，文件保存在: $OUTPUT_DIR"

PostgreSQL批量导出Python脚本

import subprocessimport osimport globfrom datetime import datetimeclass PostgreSQLExporter:"""PostgreSQL批量导出工具，支持pg_dump全库/按表/COPY三种模式"""def __init__(self, host="localhost", port=5432, user="postgres",password=None, dbname=None):self.host = hostself.port = portself.user = userself.password = passwordself.dbname = dbnameself.env = os.environ.copy()if password:self.env["PGPASSWORD"] = passworddef export_database(self, dbname: str, output_dir: str,format: str = "custom", jobs: int = 4) -> bool:"""导出整个数据库format: plain/custom/directory/tarjobs: 并行任务数（仅directory格式有效）"""os.makedirs(output_dir, exist_ok=True)timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")ext_map = {"plain": "sql", "custom": "dump", "directory": "", "tar": "tar"}fmt_flag = {"plain": "-Fp", "custom": "-Fc", "directory": "-Fd", "tar": "-Ft"}out_path = os.path.join(output_dir, f"{dbname}_{timestamp}.{ext_map[format]}")cmd = ["pg_dump", "-h", self.host, "-p", str(self.port),"-U", self.user, "-d", dbname,fmt_flag[format], "-f", out_path,"--no-owner", "--no-acl"]if format == "directory" and jobs > 1:cmd.extend(["-j", str(jobs)])result = subprocess.run(cmd, env=self.env, capture_output=True, text=True)if result.returncode == 0:size = os.path.getsize(out_path) if format != "directory" else \sum(os.path.getsize(f) for f in glob.glob(f"{out_path}/*"))print(f"  ✓ {dbname}: {size/1024/1024:.1f}MB → {out_path}")return Trueelse:print(f"  ✗ {dbname}: {result.stderr.strip()}")return Falsedef export_table_csv(self, dbname: str, schema: str, table: str,output_dir: str) -> bool:"""使用COPY命令导出单表为CSV（最快方式）"""import psycopg2os.makedirs(output_dir, exist_ok=True)out_path = os.path.join(output_dir, f"{schema}.{table}.csv")conn = psycopg2.connect(host=self.host, port=self.port, user=self.user,password=self.password, dbname=dbname)cur = conn.cursor()try:with open(out_path, "w", encoding="utf-8") as f:cur.copy_expert(f"COPY {schema}.\"{table}\" TO STDOUT "f"WITH CSV HEADER DELIMITER ',' ENCODING 'UTF8'",f)size = os.path.getsize(out_path)print(f"  ✓ {schema}.{table}: {size/1024/1024:.1f}MB → {out_path}")return Trueexcept Exception as e:print(f"  ✗ {schema}.{table}: {e}")return Falsefinally:cur.close()conn.close()def export_all_tables_csv(self, dbname: str, output_dir: str) -> dict:"""批量导出所有表为CSV"""import psycopg2conn = psycopg2.connect(host=self.host, port=self.port, user=self.user,password=self.password, dbname=dbname)cur = conn.cursor()# 获取所有用户表cur.execute("""SELECT schemaname, tablenameFROM pg_tablesWHERE schemaname NOT IN ('pg_catalog', 'information_schema')ORDER BY schemaname, tablename""")tables = cur.fetchall()cur.close()conn.close()results = {"success": 0, "failed": []}for schema, table in tables:if self.export_table_csv(dbname, schema, table, output_dir):results["success"] += 1else:results["failed"].append(f"{schema}.{table}")return results# 使用示例exporter = PostgreSQLExporter(host="localhost", user="postgres", password="your_password")# 导出整个数据库（custom压缩格式，生产环境推荐）exporter.export_database("my_database", "./pg_export", format="custom")# 导出大库（directory格式 + 4线程并行）exporter.export_database("big_database", "./pg_export_big",format="directory", jobs=4)# 批量导出所有表为CSV（数据分析用）result = exporter.export_all_tables_csv("my_database", "./pg_csv_export")print(f"完成: {result['success']}张表, 失败{len(result['failed'])}张")