数据迁移 / ETL

1. 这是什么场景 + 用户 + 挑战

场景:企业 IT / 数据团队执行一次性或周期性的数据迁移项目,把数据从老系统(ERP / CRM / 数据库 / 文件存储)搬到新系统,或把多源数据(CRM + ERP + 财务 + 第三方 SaaS)汇聚到数据仓库 / 数据湖做 ETL 处理。一次迁移项目持续数周到数月,涉及数据量从百万行到数十亿行。

典型用户:企业数据工程师、IT 架构师、BI 分析师、DBA。这些人面对的不是"写一个 Python 脚本",而是"设计一套可回滚、可校验、可审计、可重跑的迁移流水线"。他们关心三件事:数据完整性(零丢失 + 语义对齐)、可重复运行(失败可重跑,不产生副作用)、合规留痕(谁、何时、改了什么字段都有 audit)。

核心挑战(行业共识,跨多家厂商技术博客):

• Schema drift:源 / 目标 schema 在迁移期间可能漂移,字段类型 / nullable / 默认值不一致
• 字符集与时区:UTF-8 vs GBK、TIMESTAMP vs VARCHAR、null vs 空字符串,边界 case 极多
• 数据敏感:用户 PII、财务数据、医疗 PHI 不能走 SaaS 公共 endpoint,必须私有化 / VPC
• 迁移期间源系统不能停:业务不停写,数据还在涨,需要增量迁移 + 切换窗口
• 回滚预案:全量迁移到一半失败怎么办?必须能切回旧系统继续服务
• 成本失控:如果 agent 死循环调 SQL,token / API 调用 / 数据库 IO 会爆

---

2. 完整流程

2.1 数据源调研 + 映射设计

把源系统(老 ERP / 老 CRM / MySQL 主库 / S3 CSV)的表结构、字段语义、数据量级、增量方式摸清,产出 mapping 文档:源字段 → 目标字段 + 转换规则 + 校验规则。这一步通常由 DBA + 业务分析师手动出 Excel / Confluence,数据量大时是体力活。

Agent 价值:Claude Code 用 MCP 连数据库(如 @modelcontextprotocol/server-postgres 或自建 @mcp/server-bigquery)直接 SELECT column_name FROM information_schema.columns,自动生成 mapping draft,人审核确认。Manus MCP Connectors 把数据源接入做成"配置即用",GitHub / Notion / Slack / Stripe / Google Maps / REST APIs 都在列表里(manus.im/docs/integrations/mcp-connectors.md ✅)。

2.2 数据清洗 + 标准化

PII 字段脱敏(身份证 / 手机号 / 邮箱 / 卡号)、空值填充、字符集统一(GBK → UTF-8)、时区归一化(TIMESTAMP → UTC)、去重、异常值标记。这一步是 ETL 流水线最容易被忽视的环节,90% 的迁移事故来源于此。

Agent 价值:Qwen-Agent code_interpreter 在本地 Docker 容器里跑 Python pandas,边读边清洗,实时打印进度(qwenlm.github.io ✅);ChatGPT Agent hosted code interpreter 预装 pandas / numpy / scipy,Python 3.11 sandbox + 5GB 内存(platform.openai.com/docs/guides/tools-code-interpreter ✅)。

2.3 ETL 脚本编写 + 测试

把 mapping 文档翻译成 Python / SQL / Spark 脚本。脚本必须幂等(idempotent)——重复运行不产生副作用(用 upsert 而不是 insert、用 timestamp + primary key 去重)。脚本要写单元测试(边界 case:null / 空字符串 / 极大值 / 字符截断)。

Agent 价值:Devin 跑 SWE-bench Verified 风格的 end-to-end test,72% case 跑超过 10 分钟仍不放弃(cognition.com/blog/swe-bench ✅);Claude Code Stop hook 当确定性 gate,exit 2 block + stderr feed to Claude,8 次 block 强制结束,确保 typecheck + lint + test 三件套全绿(anthropic.com/engineering/claude-code-best-practices ✅)。

2.4 小批量试跑 + 数据校验

关键纪律:不要全量一次跑。先抽 1% / 1000 行 / 单分区试跑,验证 mapping 是否正确、清洗规则是否合理、性能是否在预算内。试跑后做 3 层校验:

1. Hash 校验:源 vs 目标 row hash(MD5 / SHA256),100% 一致才算过
2. Row count:源 N 行 = 目标 N 行(允许 ±0)
3. Sample 内容:随机抽 100 行人工 / LLM 比对,看语义是否对齐

Agent 价值:Claude Code BashOutput 工具读长跑 ETL 进程 log,类似 tail -f 但 agent 主循环可继续做其他事(code.claude.com ✅);ChatGPT Agent background mode background: true 异步跑 run,完成 webhook 通知 + polling(platform.openai.com/docs/guides/background ✅)。

2.5 全量迁移 + 监控

试跑通过后,分批跑全量(按日期分区 / 按 ID 范围 / 按 hash bucket),每批做 checkpoint(写 offset 到 _etl_progress 表,断点续跑)。监控 4 个维度:进度(已处理 / 总数)、速率(rows/sec)、错误率(失败行 / 总行)、延迟(源 → 目标 lag)。

Agent 价值:ChatGPT Agent Code interpreter 5GB 内存上限对百万行小数据集够用,大数据集需切片(platform.openai.com ✅);Coze 3 工作流节点监控每个节点的运行时间 / 输入 / 输出,排查哪里出问题(coze.cn/open/docs ✅);WorkBuddy 企业管理后台实时显示权限 / 成本 / 用量统计。

2.6 切换 + 回滚预案

迁移完成 + 校验全过后,双写期:新系统上线,旧系统继续写 7-14 天,两边数据对比,确认一致后才 cutover(切流量)。回滚预案:如果新系统出问题,流量切回旧系统,新系统的数据不能丢(可以暂时不写或异步回灌)。

Agent 价值:Devin session 内 "Take over" 按钮允许用户在 cutover 关键步骤接管手柄,agent 旁观(docs.devin.ai ✅);Claude Code /rewind + 5 选项(Restore code only / Restore conversation only / Restore both / Summarize from here / Summarize up to here)给"试跑失败回到上一步"的回滚能力(code.claude.com ✅)。

---

3. 需要的核心能力(链 modules)

---

4. 推荐 Agent 组合

为什么是这个组合:

• Devin + Claude Code 主力:Devin 长跑 session + Take over 负责 plan 编排和关键决策,Claude Code 负责脚本编写 + 校验 + 长跑进程管理(两者形成 plan/execute 闭环)
• Manus 辅助:MCP Connectors 把 GitHub / Notion / Stripe / SaaS 数据源"配置即用",减少手写 API 适配
• Coze 3 + WorkBuddy 辅助(国内):Coze 3 工作流编排适合定时跑批 + 非技术 stakeholder 参与;WorkBuddy 私有化四路径 + 信通院 57 项评估适合国内政企合规要求

---

5. 实战 tips

• Tip 1:先做数据 sample(1%)试跑,验证 mapping 正确 — 不要全量一次跑;mapping 错了 1% 数据 = 全量迁移失败 + 回滚 = 数天延误。试跑发现 mapping bug 成本 < 1 小时
• Tip 2:迁移脚本必须幂等(idempotent) — 支持重复运行不产生副作用。用 upsert 替 insert,用 (timestamp, primary_key) 去重,checkpoint 表存 offset。Devin CI must pass gate + Claude Code Stop hook 强制跑测试
• Tip 3:回滚预案必须先准备好,不要等失败再想 — 双写期 / cutover 步骤 / 数据回灌脚本 / DNS 切换命令全部写成 runbook,演练过再上生产
• Tip 4:数据校验做 3 层:hash + row count + sample 内容 — 单一校验不够:hash 过了但语义错(空字符串 vs null),row count 对了但内容错(GBK → UTF-8 乱码)。3 层交叉验证才稳
• Tip 5:大文件 / 大表用 Background task + BashOutput 读 log — Claude Code & 启动长跑进程,BashOutput 工具读 log(类似 tail -f),agent 主循环不被阻塞(code.claude.com ✅)
• Tip 6:敏感数据用 ZDR + 私有化部署 — ChatGPT Agent store=False API 模式(provider 不存储)或 Qwen-Agent 专属版 2.0 VPC 部署(qwenlm.github.io ✅ + aliyun.com/bailian ✅);不要把 PII / PHI 发到公共 SaaS
• Tip 7:多源并行迁移用 Multi-Agent — Claude Code sub-agent background: true + Antigravity Deep Think Parallel Reasoning(多个假设路径并发跑),比串行迁移快 3-5x(code.claude.com ✅)
• Tip 8:监控 4 维度:进度 / 速率 / 错误率 / 延迟 — 任何一项异常立刻告警;Coze 3 节点监控 / ChatGPT Agent background mode + webhook(platform.openai.com ✅)

---

6. 常见坑

• 坑 2:数据敏感(PII / PHI / 财务)必须加密 + 私有化 — 不能用 SaaS 公共 endpoint;Agentforce Einstein Trust Layer(ZDR + 毒性检测 + 动态接地)是行业答案,WorkBuddy QClaw 本地离线 / Lighthouse 云端是国产化路径(salesforce.com/agentforce ✅ + codebuddy.cn/work/ )
• 坑 3:字符集 / 时区 / null 处理容易遗漏 — GBK vs UTF-8 乱码、TIMESTAMP vs VARCHAR 时区错乱、空字符串 vs null 行为不一致。Coze 3 节点级 IF-ELSE 处理异常流程(coze.cn ✅);Antigravity 三阶段状态机 + Walkthrough Report 截图存证(blog.google/products/gemini/gemini-3/ ✅)
• 坑 4:源 / 目标 schema drift 必须 schema diff 工具 — 迁移期间源系统可能加字段,目标 schema 必须 diff 校验;Claude Code MCP 连数据库,自动比对 information_schema(04-tool.md §2.2 ✅)
• 坑 5:迁移期间源系统不能写,避免数据不一致 — 双写期或 read-only 锁定;cutover 时 DNS 切换要原子(用 Lua 脚本或专业流量切换工具)
• 坑 6:agent 死循环调 SQL / 烧 token — Claude Code Stop hook 8 次 block 强制结束;ChatGPT Agent approval_timeout: 600 配置;WorkBuddy 实测 262K token 上限触发熔断 + 通道失效(用户可能误以为还在 Auto 跑(code.claude.com)

---

7. 延伸阅读

• 模块 2:触发与自动化 → /docs/ai-agent/02-trigger(Background task + Routine + Cron 全套)
• 模块 5:执行与沙箱 → /docs/ai-agent/05-sandbox(sandbox + permission model + ZDR)
• 模块 7:验证与监控 → /docs/ai-agent/07-verification(Test + Monitoring + Audit log)
• 模块 10:治理与合规 → /docs/ai-agent/10-governance(私有化 + 加密 + 合规认证)
• Claude Code sandboxing:https://code.claude.com/docs/en/sandboxing ✅
• Claude Code background tasks:https://www.anthropic.com/news/enabling-claude-code-to-work-more-autonomously ✅
• ChatGPT Agent background mode:https://platform.openai.com/docs/guides/background ✅
• ChatGPT Agent code interpreter:https://platform.openai.com/docs/guides/tools-code-interpreter ✅
• Devin SWE-bench:https://cognition.com/blog/swe-bench ✅
• Qwen-Agent code interpreter + MCP:https://qwenlm.github.io/Qwen-Agent/en/guide/core_moduls/mcp/ ✅
• Manus MCP Connectors:https://manus.im/docs/integrations/mcp-connectors.md ✅
• Coze 3 工作流:https://www.coze.cn/open/docs/guides/workflow ✅
• WorkBuddy 私有化部署:https://www.codebuddy.cn/work/

---