怎么用脚本转换数据库表

wen 实用脚本 2026-07-31 1

用脚本高效转换数据库表结构的实战指南（附Python/Shell脚本案例）

目录导读

为什么你需要脚本化数据库表转换？ —— 手动操作的痛点与自动化收益
转换前的“体检”清单 —— 元数据采集、依赖关系与风险预判
核心武器：Python + SQLAlchemy + Pandas —— 跨库迁移的黄金组合
实战脚本拆解 —— 从MySQL到PostgreSQL的字段类型映射与数据搬运
Shell脚本的妙用 —— 定时任务、批量处理与日志告警
常见坑位与避雷指南 —— 编码、精度、自增主键与索引重建
FAQ高频问答 —— 解决你转换路上的最后一道坎

为什么你需要脚本化数据库表转换？

假设你是一家SaaS公司的数据工程师,业务快速增长，需要把核心订单表从MySQL（InnoDB）迁移到PostgreSQL（以利用其JSONB和窗口函数），如果你手动执行：

怎么用脚本转换数据库表

创建新表结构（约30分钟）
逐字段比对类型（如TINYINT → SMALLINT，VARCHAR(255) → TEXT）
编写ETL脚本搬运1000万行数据（约2小时）
重建索引、触发器、外键（约40分钟）

这还是一次性的,如果遇到主从切换、分库分表合并，工作量呈指数级上升，而脚本化转换能将整个过程压缩到一条命令行，输出结构化日志，失败可回滚，且可重复执行。

核心收益：

可重复性：同一脚本可应用于测试、预发、生产环境。
可审计性：每一次转换都有完整日志（源表行数、目标表行数、失败行数）。
类型自动映射：避免手抖把DECIMAL(10,2)写成REAL导致精度丢失。

转换前的“体检”清单

在写脚本前,你必须回答以下问题（否则脚本会变成“事故现场”）：

1 元数据采集

使用SQL查询源库的information_schema：

SELECT table_name, column_name, data_type, character_maximum_length, is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = 'your_schema' AND table_name = 'orders';

同样查询索引（SHOW INDEX FROM orders）、外键、触发器。

2 依赖关系图谱

是否有视图、存储过程引用该表？
是否有应用代码硬编码了列名或类型（如Python中INT转str）？
是否有夜间批处理任务依赖该表结构？

3 风险预判

数据量级：<100万行可全量；>500万行需要分片（WHERE id % N = M）或使用SELECT * FROM t WHERE id > ?流式处理。
停机窗口：业务允许多久不可写？是否要采用双写或CDC（Change Data Capture）方案？
字符集与排序规则：MySQL的utf8mb4对应PostgreSQL的UTF8，但utf8_bin与C排序规则不一致会导致索引失效。

核心武器：Python + SQLAlchemy + Pandas

1 为什么选这三个？

SQLAlchemy：它提供了一个高抽象的MetaData对象，可以反射源库表结构，并生成目标库的DDL，它帮你处理了TEXT、DATETIME等跨方言差异。
Pandas：适合处理中等规模数据集（<200万行），提供to_sql方法快速批量插入，但注意：Pandas会占较大内存，超大规模请用psycopg2的copy_expert。
Python：轻松处理异常、重试、进度条（tqdm）。

2 脚本骨架（简化版）

import sqlalchemy as sa
import pandas as pd
from sqlalchemy.schema import CreateTable
# 1. 连接源库和目标库
src_engine = sa.create_engine('mysql+pymysql://user:pass@host:3306/db')
dst_engine = sa.create_engine('postgresql+psycopg2://user:pass@host:5432/db')
# 2. 反射源表元数据
src_metadata = sa.MetaData(bind=src_engine)
src_table = sa.Table('orders', src_metadata, autoload_with=src_engine)
# 3. 自动生成目标表DDL（转换类型）
ddl = CreateTable(src_table, bind=dst_engine).compile(dialect=dst_engine.dialect)
# 手动微调：例如将src_table的TINYINT映射为SMALLINT
# 4. 流式读取源数据
offset = 0
while True:
    sql = f"SELECT * FROM orders LIMIT 50000 OFFSET {offset}"
    df = pd.read_sql(sql, src_engine)
    if df.empty: break
    # 处理空值、时间戳转字符串等
    df.to_sql('orders', dst_engine, if_exists='append', index=False)
    offset += len(df)

关键改进点：

使用LIMIT/OFFSET在大表上性能差，改用WHERE id > last_id的键集分页（Seek Method）。
目标表创建后,务必先导入数据再创建索引，避免每次INSERT维护索引。
使用multiprocessing或concurrent.futures并行读取不同ID段。

实战脚本拆解：MySQL → PostgreSQL 字段类型映射

这是脚本的灵魂,以下映射表直接参考官方文档和社区经验：

MySQL	PostgreSQL	注意
`TINYINT(1)`	`BOOLEAN`	若存0/1，可直接转换；若存-128~127，用`SMALLINT`
`DATETIME`	`TIMESTAMP WITHOUT TIME ZONE`	注意时区。`DATETIME`不含时区，`TIMESTAMPTZ`含时区
`MEDIUMINT`	`INTEGER`	PostgreSQL无MEDIUMINT，用INT或扩展至BIGINT
`VARCHAR(255)`	`VARCHAR(255)`	长度一致，但注意字符集差异（utf8mb4下255字符=1020字节，PG中varchar按字符计）
`BLOB`	`BYTEA`	直接映射，但写入时需`psycopg2.Binary()`
`JSON`	`JSONB`	业务上建议选择JSONB，支持索引和高效查询
`UNSIGNED INT`	`BIGINT`	MySQL无符号范围0~4294967295，PG的INT有符号到2147483647，必须提升到BIGINT

自动生成映射器的技巧：

TYPE_MAP = {
  'TINYINT': 'SMALLINT',
  'MEDIUMINT': 'INTEGER',
  'DATETIME': 'TIMESTAMP WITHOUT TIME ZONE',
  'BLOB': 'BYTEA',
  # ... 更多映射
}
def convert_type(column):
    if column.type.python_type == int and column.unsigned:
        return sa.types.BIGINT()
    return TYPE_MAP.get(str(column.type).split('(')[0].upper(), column.type)

Shell脚本的妙用：自动化与监控

Python负责数据搬运,Shell负责编排。

1 定时批处理（crontab）

# 每天晚上2点执行转换，记录日志
0 2 * * * cd /data/script && python convert.py >> logs/convert_$(date +\%F).log 2>&1

2 带重试与告警的封装

#!/bin/bash
set -e  # 出错即停
PIPELINE_ID="orders_mysql2pg_$(date +%s)"
python preflight_check.py --pipeline $PIPELINE_ID  # 检查源表数据量是否突变
if [ $? -eq 0 ]; then
  python transfer.py --pipeline $PIPELINE_ID || {
    echo "转换失败，触发回滚脚本"
    python rollback.py --pipeline $PIPELINE_ID
    curl -X POST -d "pipeline=$PIPELINE_ID&status=failed" https://alerts.yourcompany.com/webhook
    exit 1
  }
else
  echo "前置检查异常，跳过转换"
fi

常见坑位与避雷指南

1 字符集乱码

现象：中文变。
原因：源库连接未指定charset='utf8mb4'，或目标库的client_encoding错误。
解方：连接串加?charset=utf8mb4；PG侧设置SET NAMES 'UTF8'；批量写入时确保psycopg2使用utf-8。

2 精度丢失

现象：DECIMAL(20,2) → PG NUMERIC正确，但若错误映射为DOUBLE PRECISION，1.10会变成1.0999999。
解方：强制映射为NUMERIC，并在JDBC/psycopg2中设置pg8000的decimal返回。

3 自增主键坑

MySQL AUTO_INCREMENT → PG SERIAL，但SERIAL用INTEGER，若MySQL是BIGINT UNSIGNED，必须用BIGSERIAL。
插入时不要复制主键值，直接插入NULL，让PG分配，否则会与sequence不同步，导致后续插入冲突。

4 索引重建策略

先建表或IF NOT EXISTS，不建索引。
导入数据后,用原生SQL创建索引：CREATE INDEX CONCURRENTLY idx_orders_created ON orders (created_at);（PG支持并发，不锁表）。
MySQL侧需要ALTER TABLE ... ADD INDEX会锁表，建议使用gh-ost或pt-osc工具，但这里是目标库，简单执行即可。

FAQ高频问答

Q1：我的表有大量BLOB字段，Pandas真的能处理吗？

可以,但内存爆炸，建议用psycopg2的copy_expert配合BytesIO流式写入：
cur.copy_expert("COPY orders (id, data) FROM STDIN WITH (FORMAT BINARY)", file_like_obj)
或者改用MySQL的SELECT ... INTO OUTFILE，再\copy到PG。

Q2：转换过程中，源表数据还在更新怎么办？

方案A：先锁表（LOCK TABLES orders WRITE），转换完解锁——但这意味着停机。方案B：用CDC工具（Debezium监听binlog），转换结束后追平增量——适合7x24业务。建议：根据业务容忍度选择，脚本中加--allow_dirty_read参数控制。

Q3：如何验证转换后数据完全一致？

写一个校验脚本：

-- 对比行数
SELECT COUNT(*) FROM mysql_db.orders;
SELECT COUNT(*) FROM pg_db.orders;
-- 对比每列MD5（注意排序）
CHECKSUM TABLE mysql_db.orders;  -- MySQL专用，返回全局校验和
-- PG侧用`pg_checksums`或逐行求MD5后聚合。

Q4：脚本跑了一半，内存溢出怎么办？

不要再扩大内存,改成分片+游标：

with src_engine.connect().execution_options(stream_results=True) as conn:
    result = conn.exec_driver_sql("SELECT * FROM orders")
    while True:
        chunk = result.fetchmany(10000)
        if not chunk: break
        df = pd.DataFrame(chunk, columns=result.keys())
        df.to_sql('orders', dst_engine, if_exists='append', index=False)

Q5：有没有现成的开源工具，不写代码？

有,但都有限制：

pgloader：支持MySQL→PG，但类型映射有限，复杂逻辑得改脚本。

AWS DMS：适合云迁移，但无法处理自定义业务逻辑。

自己写脚本最灵活,加上argparse，可以做成“转换即配置”平台。

最后总结：脚本转换数据库表，本质是数据桥梁——用SQLAlchemy反射结构，用Pandas/游标搬运数据，用Shell保障调度，用日志和校验兜底。不要试图一次写完美，建议先在测试库模拟全量数据，比对每行校验值，再灰度到生产，掌握此方法论，任何数据库之间的表转换对你来说都是“模板+微调”，如果你在转换中遇到特定数据库（Oracle/DB2/SQL Server）的怪癖，欢迎留言交流。