1. 停机迁移的真实场景与痛点拆解如果你正在维护一个跑了三五年的 Web 项目数据库还是 MySQL 5.7那你大概率已经遇到过这些情况复杂窗口函数写起来别扭、JSON 字段查询性能拉胯、地理空间索引基本没法用。业务量一上来慢查询日志一天能刷几百兆。这时候把库迁到 PostgreSQL 16 就成了一个绕不开的选项。但迁移这件事最怕的不是迁不过去而是迁过去之后数据对不上。我见过太多团队在停机窗口里手忙脚乱pgloader 跑完了业务切过去第二天发现某张订单表的金额精度丢了两位或者自增主键从 1 开始重新计数导致主键冲突。这类问题一旦在生产环境暴露回滚成本极高。所以这篇内容的核心目标很明确在一个可控的停机窗口内完成 MySQL 5.7 到 PostgreSQL 16 的 schema 转换、全量数据迁移、一致性校验并且准备好回滚预案。整个流程用 Cursor 辅助生成迁移脚本和校验代码把重复劳动交给 AI把判断力留给自己。适合谁看后端开发、DBA、运维工程师尤其是那些手头有几十 GB 到几百 GB 数据、停机窗口只有一两个小时的团队。如果你只是本地玩票迁移几万行数据这篇的方案同样适用只是可以跳过部分并行调优参数。迁移的本质不是复制数据而是在目标库重建一套语义等价的结构并证明数据没有失真。MySQL 的datetime允许0000-00-00 00:00:00这种脏值PostgreSQL 直接拒绝写入MySQL 的AUTO_INCREMENT和 PostgreSQL 的SEQUENCE是两套完全不同的机制。这些差异如果不在迁移前处理就会在割接当晚变成一个个报错。我试过的方案里pgloader 是目前综合成本最低的选择。它基于 Common Lisp 实现利用 PostgreSQL 的 COPY 流协议做多线程导入几十万行每秒的吞吐是常态。更重要的是它能自动处理类型映射、重置序列、生成索引还能把出错的行隔离到单独日志里而不中断整个迁移。相比之下手写 Python 脚本导入在大表面前基本没有吞吐优势Navicat 这类图形工具在 Linux 无人值守场景下也不方便。下面这张表是我实际对比后的结论你可以直接参考方案维度pgloader 自动同步自研 Python 脚本Navicat 图形工具数据吞吐量极高多线程 COPY 流协议较低单线程批量插入中等依赖客户端缓存Schema 自动适配强内置类型转换与序列重置差需手写 DDL较强但序列常漏断点恢复强错误行隔离不中断差需自己写重试差连接抖动即崩溃方案成本零开源高开发调试工时大中等需商业授权选型确定后接下来的问题就是怎么让 Cursor 帮我们把迁移配置和校验脚本快速写出来同时把 TaoToken 的统一 Key 接进校验流程让整个迁移过程可观测、可复现。2. TaoToken 统一 Key 前置配置与 Cursor 提示词模板在正式写迁移脚本之前先把工具链准备好。这里要解决两个问题一是 Cursor 里怎么高效生成迁移相关的代码二是迁移校验脚本如果需要调用模型能力比如自动分析报错日志、生成字段映射建议怎么用一个统一的 Key 来管理。TaoToken 在这里的角色是统一模型接入层。你不需要在 Cursor、校验脚本、日志分析工具里分别配置不同的模型 Key而是通过一个 Base URL 和一把 Key 覆盖所有场景。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。先说 Cursor 侧的配置。打开 Cursor 设置找到模型配置区域填入以下三件套Base URLhttps://taotoken.net/apiAPI Key在控制台创建的 Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewriteModel ID根据你的套餐选择比如claude-sonnet-4-20250514或gpt-4o配置完成后你可以在 Cursor 里直接用自然语言让它生成 pgloader 的.load配置文件。这里给一个我常用的提示词模板你可以直接复制你是一个数据库迁移专家。请帮我生成一个 pgloader 配置文件用于将 MySQL 5.7 的 quant_trade_db 迁移到 PostgreSQL 16。 要求 1. 源库连接mysql://user:pass127.0.0.1:3306/quant_trade_db 2. 目标库连接postgresql://postgres:pass127.0.0.1:5432/quant_trade_db 3. 开启 create tables、create indexes、reset sequences、foreign keys 4. 表名和字段名统一转小写 5. workers4concurrency2batch rows25000 6. 将 datetime 转为 timestamptztinyint 转为 boolean 7. 排除以 temp_ 开头的表 请输出完整的 .load 文件内容并加上中文注释说明每个参数的作用。这个提示词的关键在于把连接信息、转换规则、性能参数都写清楚Cursor 生成的配置基本可以直接用。如果你用的是 Claude Code 或者 Cline 这类工具配置方式类似核心都是 Base URL Key Model ID 三件套。对于校验脚本如果你希望它具备自动分析迁移报错的能力可以在 Python 脚本里通过 OpenAI 兼容接口调用 TaoToken。配置片段如下import openai client openai.OpenAI( base_urlhttps://taotoken.net/api, api_key你的_TaoToken_Key ) def analyze_migration_error(error_log: str) - str: response client.chat.completions.create( modelclaude-sonnet-4-20250514, messages[ {role: system, content: 你是数据库迁移排错专家请分析以下 pgloader 报错日志并给出修复建议。}, {role: user, content: error_log} ] ) return response.choices[0].message.content这样配置的好处是迁移过程中如果 pgloader 报出类型转换错误你可以直接把日志丢给模型分析不用自己逐行啃文档。模型对话入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。需要提醒的是校验脚本里调用模型是可选增强不是迁移的必需步骤。核心的行数校验和 MD5 指纹校验用纯 SQL 和 Python 就能完成模型只负责帮你解读报错、生成修复 SQL 这类辅助工作。不要把生产库的连接信息直接发给模型脱敏后再分析。环境依赖方面在中转服务器上装好这些pip install pgloader3.6.9 pip install psycopg2-binary2.9.7 pip install mysql-connector-python8.0.33 pip install openai1.30.0pgloader 本身是通过系统包管理器安装的Ubuntu 下用apt-get install pgloader即可。Python 依赖用 pip 装。中转服务器建议和 PostgreSQL 部署在同一内网千兆带宽下迁移几百 GB 数据的时间可以压缩到分钟级。3. 可复制的迁移配置与校验脚本这一节是整篇的核心所有配置和代码都可以直接复制使用。我会把 pgloader 的.load文件、Python 校验脚本、Sequence 修复 SQL 分开讲清楚。3.1 pgloader 迁移定义文件 migration.load在迁移中转服务器上创建migration.load内容如下/* -*- coding: utf-8 -*- */ LOAD DATABASE FROM mysql://quant_root:mysql_pass127.0.0.1:3306/quant_trade_db INTO postgresql://postgres:pg_pass127.0.0.1:5432/quant_trade_db WITH include no drop, create tables, create indexes, reset sequences, foreign keys, lowercase identifiers, workers 4, concurrency 2, batch rows 25000 CAST type datetime to timestamptz drop default drop not null, type date to date drop default drop not null, type tinyint to boolean using tinyint-to-boolean, type double to double precision EXCLUDING TABLE NAMES MATCHING temp_ BEFORE LOAD DO $$ create schema if not exists public; $$;逐行解释关键参数。include no drop表示目标库里已存在但不冲突的表不动它避免误删。create tables和create indexes让 pgloader 根据 MySQL 的 schema 自动在 PostgreSQL 建表建索引。reset sequences是最关键的一项它会在数据导入后自动把 PostgreSQL 的 SEQUENCE 重置到表中最大 ID防止业务上线后主键冲突。lowercase identifiers强制表名字段名转小写因为 PostgreSQL 默认对未加引号的标识符做小写处理而 MySQL 在 Linux 下大小写敏感不统一会导致查询找不到表。CAST部分是类型映射规则。datetime to timestamptz把 MySQL 的 datetime 转成带时区的时间戳drop default drop not null去掉默认值和非空约束避免脏数据导致整批拒绝。tinyint to boolean处理那些用 0/1 表示布尔值的字段。double to double precision保证浮点精度。EXCLUDING TABLE NAMES MATCHING temp_排除临时表这些表通常不需要迁移。3.2 数据一致性校验脚本 check_consistency.py迁移完成后必须验证数据没有失真。这个脚本做两件事行数对齐校验和 MD5 指纹校验。# -*- coding: utf-8 -*- import mysql.connector import psycopg2 import hashlib import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(migration.validator) MYSQL_CONFIG { host: 127.0.0.1, port: 3306, user: quant_root, password: mysql_pass, database: quant_trade_db } PG_CONFIG { host: 127.0.0.1, port: 5432, user: postgres, password: pg_pass, database: quant_trade_db } TABLES_TO_CHECK [tb_user_accounts, tb_trader_positions, tb_order_history] def get_mysql_connection(): return mysql.connector.connect(**MYSQL_CONFIG) def get_pg_connection(): return psycopg2.connect(**PG_CONFIG) def check_row_counts(mysql_conn, pg_conn): mysql_cursor mysql_conn.cursor() pg_cursor pg_conn.cursor() is_all_consistent True for table_name in TABLES_TO_CHECK: mysql_cursor.execute(fSELECT COUNT(*) FROM {table_name}) mysql_count mysql_cursor.fetchone()[0] pg_table table_name.lower() pg_cursor.execute(fSELECT COUNT(*) FROM {pg_table}) pg_count pg_cursor.fetchone()[0] if mysql_count pg_count: logger.info(f表 {table_name} 行数一致! MySQL: {mysql_count} | PG: {pg_count}) else: logger.error(f表 {table_name} 行数不一致! MySQL: {mysql_count} | PG: {pg_count}) is_all_consistent False mysql_cursor.close() pg_cursor.close() return is_all_consistent def check_data_md5(mysql_conn, pg_conn): mysql_cursor mysql_conn.cursor() pg_cursor pg_conn.cursor() logger.info(开始核心表 MD5 指纹校验...) mysql_sql SELECT CONCAT(IFNULL(id, ), _, IFNULL(price, ), _, IFNULL(amount, ), _, IFNULL(status, )) FROM tb_order_history ORDER BY id pg_sql SELECT concat(coalesce(id::text, ), _, coalesce(price::text, ), _, coalesce(amount::text, ), _, coalesce(status, )) FROM tb_order_history ORDER BY id mysql_cursor.execute(mysql_sql) mysql_hasher hashlib.md5() while True: rows mysql_cursor.fetchmany(10000) if not rows: break for row in rows: mysql_hasher.update(str(row[0]).encode(utf-8)) mysql_md5 mysql_hasher.hexdigest() pg_cursor.execute(pg_sql) pg_hasher hashlib.md5() while True: rows pg_cursor.fetchmany(10000) if not rows: break for row in rows: pg_hasher.update(str(row[0]).encode(utf-8)) pg_md5 pg_hasher.hexdigest() mysql_cursor.close() pg_cursor.close() if mysql_md5 pg_md5: logger.info(f核心数据指纹校验成功! MD5: {mysql_md5}) return True else: logger.critical(f数据不一致! MySQL-MD5: {mysql_md5} | PG-MD5: {pg_md5}) return False def main(): try: mysql_conn get_mysql_connection() pg_conn get_pg_connection() counts_ok check_row_counts(mysql_conn, pg_conn) md5_ok check_data_md5(mysql_conn, pg_conn) if counts_ok and md5_ok: logger.info(迁移后一致性校验全部通过可以割接。) else: logger.error(一致性校验未通过请排查后再割接。) mysql_conn.close() pg_conn.close() except Exception as e: logger.critical(f校验异常: {str(e)}) if __name__ __main__: main()这个脚本的核心思路是把每行数据的关键字段拼成一个字符串按主键排序后逐行喂给 MD5 哈希器最后比对两端的哈希值。只要有一行数据不一致MD5 就会完全不同。分批读取是为了避免大表一次性加载到内存。3.3 Sequence 修复 SQL即使 pgloader 开了reset sequences某些情况下序列仍可能落后于实际最大 ID。迁移完成后建议手动跑一遍这个 SQL 兜底SELECT setval( pg_get_serial_sequence(tb_order_history, id), coalesce(max(id), 1) ) FROM tb_order_history;对每张有自增主键的表都执行一次确保业务上线后插入新数据不会主键冲突。4. 停机窗口内的执行与验证结果配置和脚本准备好之后真正的割接在停机窗口内执行。整个流程分四步锁定源库、执行迁移、运行校验、切换流量。4.1 锁定 MySQL 源库在割接窗口开始时先停止所有向 MySQL 写入的上游服务。然后登录 MySQL 执行只读锁定FLUSH TABLES WITH READ LOCK; SET GLOBAL read_only ON;这两条命令的作用是防止迁移过程中产生增量脏数据。FLUSH TABLES WITH READ LOCK会阻塞所有写操作并刷新表缓存read_only ON确保后续连接也只能读。注意这个锁在迁移期间一直持有所以停机窗口要预留足够时间。4.2 执行 pgloader 迁移在中转服务器上运行pgloader migration.loadpgloader 会输出多线程进度报表每张表的读取行数、导入行数、错误数、耗时都会实时打印。一个典型的输出如下table name errors read imported time ------------------- -------- -------- ---------- -------- before load do 0 1 1 0.102s tb_user_accounts 0 15000 15000 1.890s tb_order_history 0 1200000 1200000 12.450s tb_trader_positions 0 58000 58000 2.102s reset sequences 0 12 12 1.560s ------------------- -------- -------- ---------- -------- Total import time 0 1273000 1273000 18.104s重点看errors列如果某张表有错误行pgloader 会把它们写到单独的.log文件里迁移本身不会中断。迁移完成后去检查这些日志确认错误行是否可接受。4.3 运行一致性校验执行校验脚本python3 check_consistency.py预期输出INFO - 表 tb_user_accounts 行数一致! MySQL: 15000 | PG: 15000 INFO - 表 tb_order_history 行数一致! MySQL: 1200000 | PG: 1200000 INFO - 表 tb_trader_positions 行数一致! MySQL: 58000 | PG: 58000 INFO - 开始核心表 MD5 指纹校验... INFO - 核心数据指纹校验成功! MD5: 8f4c09d57a2e8c23d149023efd688cf2 INFO - 迁移后一致性校验全部通过可以割接。看到 MD5 一致说明数据在迁移过程中没有发生截断、精度丢失或编码错误。这时候可以放心切换流量。4.4 切换流量与解除锁定把应用配置里的数据库连接从 MySQL 改成 PostgreSQL重启服务。确认业务正常后在 MySQL 上解除只读UNLOCK TABLES; SET GLOBAL read_only OFF;到这里一次完整的停机迁移就完成了。整个过程在数据量 120 万行的测试环境下从锁定到校验通过大约 20 秒。生产环境数据量更大但 pgloader 的多线程吞吐能力足以把时间控制在可接受的窗口内。如果你在迁移过程中需要模型辅助分析报错可以通过 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 管理你的 Key确保校验脚本和 Cursor 用的是同一套凭证。5. 迁移报错排查与回滚预案即使准备充分迁移过程中仍可能遇到报错。这一节列出最常见的几类问题和对应的修复动作。5.1 datetime field overflow报错现象pgloader 日志里出现Database error: datetime field overflow整批数据被拒绝。原因MySQL 允许0000-00-00 00:00:00这种零值日期PostgreSQL 的 timestamp 类型直接拒绝。修复在迁移前清洗源数据UPDATE tb_order_history SET update_time 1970-01-01 00:00:00 WHERE update_time 0000-00-00 00:00:00;或者在 pgloader 的 CAST 规则里加转换函数把零值映射为合法日期。推荐前者因为清洗后的数据在目标库语义更清晰。5.2 Duplicate key value violates unique constraint报错现象业务切换到 PostgreSQL 后插入新数据时报主键冲突。原因PostgreSQL 的 SEQUENCE 当前值小于表中实际最大 ID。pgloader 的reset sequences偶尔会漏掉某些表尤其是迁移过程中有手动跳号的情况。修复对每张有自增主键的表执行SELECT setval( pg_get_serial_sequence(表名, id), coalesce(max(id), 1) ) FROM 表名;5.3 local proxy failed / 401 报错报错现象校验脚本调用模型接口时返回401 Unauthorized或local proxy failed。原因Base URL 或 API Key 配置错误。常见的是 Base URL 多写了路径或者 Key 复制时带了空格。修复确认 Base URL 是https://taotoken.net/api不要加/v1或其他后缀。Key 从控制台重新复制确保没有首尾空格。如果用的是 Cursor检查模型配置里的三件套是否完整Base URL、API Key、Model ID。5.4 reading choices 报错报错现象模型返回结果解析失败日志里出现reading choices相关错误。原因通常是请求格式不兼容或者模型 ID 写错了。修复确认 Model ID 和你的套餐匹配。接入文档里有完整的模型列表和请求示例地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。如果用的是 OpenAI SDK确保base_url参数正确传入。5.5 回滚预案如果一致性校验发现 MD5 不一致说明数据发生了严重失真必须立即回滚。回滚步骤第一步把应用配置的数据库连接改回 MySQL重启服务。第二步在 MySQL 上解除只读UNLOCK TABLES; SET GLOBAL read_only OFF;第三步检查 pgloader 的错误日志文件定位是哪张表、哪些行出了问题。常见原因是字符集不匹配或类型转换规则遗漏。第四步修复.load配置或清洗源数据后重新申请割接窗口再迁一次。回滚的关键是快。因为 MySQL 一直处于只读状态数据没有变化切回去之后业务可以立即恢复。这也是为什么停机迁移方案要保留源库不动直到校验完全通过。6. 把 TaoToken 接入迁移校验的完整动作最后说一下 TaoToken 在整个迁移流程里的具体接入方式。它不是迁移的必需组件但能让你的排错和校验效率提升一个档次。场景一Cursor 里生成迁移脚本。在 Cursor 设置里配置 Base URL 为https://taotoken.net/api填入 API Key选择 Model ID。然后用前面给的提示词模板让 Cursor 生成.load文件和校验脚本。这样你不需要在多个工具之间切换 Key一套凭证覆盖所有编码场景。场景二校验脚本自动分析报错。在check_consistency.py里加入模型调用逻辑当 MD5 不一致时把差异行的样本和 pgloader 错误日志发给模型让它给出可能的原因和修复 SQL。配置片段import openai client openai.OpenAI( base_urlhttps://taotoken.net/api, api_key你的_TaoToken_Key ) def analyze_diff(error_log: str) - str: resp client.chat.completions.create( modelclaude-sonnet-4-20250514, messages[ {role: system, content: 分析数据库迁移差异给出修复建议。}, {role: user, content: error_log} ] ) return resp.choices[0].message.content场景三长期迁移任务的 Agent 化。如果你需要频繁做数据库迁移比如多环境同步、多租户割接可以考虑用 Coding Plan 把整个流程 Agent 化。入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 适合需要长期、批量执行迁移任务的团队。接入验证的动作很简单配置完成后在 Cursor 里发一条测试请求或者在 Python 里跑一次client.chat.completions.create能正常返回就说明三件套配置正确。如果报 401检查 Key如果报连接失败检查 Base URL如果报模型不存在检查 Model ID。整个迁移方案的核心逻辑是用 pgloader 做重活用 Python 做校验用 TaoToken 做辅助分析。三者各司其职把停机窗口内的不确定性降到最低。数据迁移这件事宁可多花十分钟校验也不要带着隐患上线。