| name | db-ferry |
| description | 数据库迁移配置生成与执行指导。当用户提到数据库迁移、数据搬运、db-ferry、或需要在不同数据库之间移动数据时触发。帮助创建 task.toml 配置文件、生成正确的迁移命令、提示数据丢失风险。 |
db-ferry 数据迁移 Skill
触发场景
- 用户提及 "数据库迁移"、"数据搬运"、"db-ferry"
- 需要在不同数据库之间移动/同步数据
- 用户提供了一个
task.toml 文件需要修改或排查
- 询问如何将某个数据库的数据迁移到另一个数据库
快速概览
db-ferry 是一个声明式 CLI 工具（Go 编写），通过 TOML 配置文件定义迁移任务，支持 Oracle/MySQL/PostgreSQL/SQL Server/SQLite/DuckDB 之间的数据搬运。工具自动创建目标表结构、批量插入、可选创建索引、支持进度条显示。独立任务按 DAG 自动并行执行。0.9.0 新增 Prometheus/OTLP 指标导出、数据质量断言、守护进程模式、Lua/JS 行级插件、S3/GCS DLQ、跨库内存 JOIN、Webhook 通知、SSE 实时进度流等能力。
配置文件生成流程
Step 1: 理解用户需求
确认以下信息：
- 源数据库：类型、连接信息、需要迁移的表/SQL 查询
- 目标数据库：类型、连接信息、目标表名
- 迁移范围：全表/部分字段/条件过滤
- 写入模式：替换(默认)/追加/合并(upsert)
- 是否有索引需求
Step 2: 生成 [[databases]] 配置
每个数据库需要一个唯一 name 作为别名。各类型必填字段如下：
| 类型 | 必填字段 | 选填字段 |
|---|
oracle | name, type, host, service, user, password | port (默认1521) |
mysql | name, type, host, database, user, password | port (默认3306) |
postgresql | name, type, host, database, user, password | port (默认5432) |
sqlserver | name, type, host, database, user, password | port (默认1433) |
sqlite | name, type, path | — |
duckdb | name, type, path | 支持 :memory: |
注意：SQLite 和 DuckDB 只需要 path，不需要 host/port/user/password。
连接池与副本（可选）：
| 字段 | 说明 |
|---|
pool_max_open | 连接池最大打开连接数 |
pool_max_idle | 连接池最大空闲连接数 |
replica_fallback | 副本不可用时是否回退到主库 |
[[databases.replicas]] | 只读副本，含 host（可选 port）和 priority |
TLS/SSL（可选）：
| 字段 | 说明 |
|---|
ssl_mode | disable / require / verify-ca / verify-full |
ssl_cert | 客户端证书路径 |
ssl_key | 客户端私钥路径 |
ssl_root_cert | CA 根证书路径 |
Step 3: 生成 [[tasks]] 配置
每个任务必填字段：
| 字段 | 说明 |
|---|
table_name | 目标表名（迁移前会自动 drop/create） |
sql | 源库 SQL 查询语句 |
source_db | 引用 [[databases]] 中的别名 |
target_db | 引用 [[databases]] 中的别名 |
可选字段：
| 字段 | 说明 | 默认值 |
|---|
ignore | 设为 true 跳过此任务 | false |
mode | 写入模式: replace/append/merge/upsert | replace |
batch_size | 每批插入行数 | 1000 |
max_retries | 批量插入失败重试次数 | 0 |
validate | 迁移后校验: none/row_count/checksum/sample | none |
merge_keys | merge/upsert 模式的匹配键（必填） | — |
resume_key | 增量续传的字段名 | — |
resume_from | 增量起点 SQL 字面量 | — |
state_file | 断点状态文件路径（JSON） | — |
allow_same_table | 允许源库=目标库时执行（⚠️ 风险） | false |
skip_create_table | 跳过目标表 drop/create | false |
pre_sql | 任务执行前在目标库执行的自定义 SQL | — |
post_sql | 任务执行后在目标库执行的自定义 SQL | — |
dlq_path | 死信队列输出路径，隔离失败行；支持本地路径、S3（s3://）、GCS（gs://） | — |
dlq_format | 死信队列格式: jsonl/csv | jsonl |
depends_on | 任务依赖（按 table_name 声明），支持 DAG 调度 | — |
schema_evolution | append/merge 模式下自动 ALTER TABLE ADD COLUMN | false |
columns | 列级映射与转换（source→target，可选 transform） | — |
masking | PII 脱敏规则（column、rule，可选 range/value） | — |
adaptive_batch | 自适应批量调优（enabled、min_size、max_size、target_latency_ms、memory_limit_mb） | — |
shard | 范围分片并行读取（enabled、shards），需 resume_key，仅 append/merge | — |
validate_sample_size | validate=sample 时的采样行数 | — |
[[tasks.sources]] / [tasks.join] | 跨库内存 JOIN；每个 source 需 alias、db、sql；join 需 keys 和 type（inner/left/right） | — |
[[tasks.assertions]] | 数据质量断言（column/columns、rule、on_fail）；规则：not_null/range/in_set/unique/regex/min_length/max_length | — |
[tasks.plugin] | 行级转换插件（engine: lua/javascript、script、可选 timeout_ms） | — |
Step 4: 配置索引（可选）
在 [[tasks]] 下添加 [[tasks.indexes]]：
| 字段 | 说明 |
|---|
name | 索引名称（全局唯一） |
columns | 列列表，支持排序: ["col:ASC", "col2:DESC"] |
unique | 是否唯一索引 |
where | 部分索引条件（仅 SQLite 目标支持） |
Step 5: 全局配置（Metrics / Notify / Schedule）
在根级别添加可选全局段：
Metrics 导出 [metrics]：
enabled: 是否启用指标收集listen_addr: Prometheus pull 端点地址，如 :9090endpoint: OTLP HTTP push 端点，如 http://localhost:4318/v1/metricsinterval: push 间隔，默认 30s
Webhook 通知 [notify]：
on_success: 迁移成功时调用的 Webhook URL 列表on_failure: 迁移失败时调用的 Webhook URL 列表timeout: 单次请求超时retry: 失败重试次数
Schedule 定时调度 [schedule]（daemon 模式生效）：
cron: cron 表达式或描述符，如 @every 1htimezone: IANA 时区名retry_on_failure / max_retry: 失败重试missed_catchup: 错过调度时立即补执行start_at / end_at: 执行窗口边界
Step 6: 风险检查清单
生成配置后，务必确认以下风险点：
- 同库迁移：
source_db == target_db 时必须设置 allow_same_table = true，否则配置验证会报错
- replace 模式：会先 DROP 目标表再重建，会丢失目标表原有数据
- 密码安全：
task.toml 明文存储密码，建议 chmod 600 task.toml，不要提交到版本控制
- merge_keys 一致性：mode=merge 时
merge_keys 必填，且应对应目标表的唯一约束
- 增量续传：
state_file 需配合 resume_key，resume_key 需配合 state_file 或 resume_from
- 校验模式：
validate 支持 row_count（行数对比）、checksum（哈希行级对比）、sample（随机采样校验）；merge 模式下校验自动跳过
- SQL 钩子：
pre_sql 在目标表创建后、数据插入前执行；post_sql 在数据插入完成后执行，可用于创建物化视图、刷新统计信息等
- PII 脱敏：
masking 在数据写入目标前对列值进行脱敏，支持 phone_cn、email、id_card_cn、hash 等 8 种规则
- 列映射转换：
columns 可重命名列并应用 transform 表达式（如 UPPER(source_col)），注意 transform 由目标库执行
- 自适应批量：
adaptive_batch 根据延迟和内存动态调整 batch_size，启用后 task 的 batch_size 作为初始值
- 分片并行：
shard 将单表按 resume_key 范围拆分为多片并行读取，仅支持 append/merge 模式，不支持 state_file
- Schema 演进：
schema_evolution 在 append/merge 模式下检测到源端新增列时自动 ALTER TABLE ADD COLUMN
- 迁移审计：
history.enabled 会在目标库自动创建审计表记录每次迁移
- Diff 对比：
db-ferry diff 需任务已执行过且目标表存在，默认输出 JSON 格式差异
- 数据质量断言：
assertions 在数据写入目标前对行/列进行规则校验，on_fail=abort 会终止任务，warn 仅记录，dlq 将失败行写入死信队列
- 行级插件：
plugin 在每行数据写入目标前执行 Lua/JavaScript 脚本转换，注意脚本执行性能开销
- 跨库 JOIN：
sources + join 将多个数据库数据在内存中 JOIN 后写入目标，不支持断点续传、分片或 CDC
- SSE 实时进度：启动时传入
-sse-port :8080 可在 /events 订阅实时任务进度，/status 查看当前状态
CLI 命令参考
| 命令 | 说明 |
|---|
db-ferry | 使用当前目录 task.toml 执行迁移 |
db-ferry -config <path> | 指定配置文件路径 |
db-ferry -v | 详细日志输出（调试用） |
db-ferry config init | 交互式配置向导，引导选择引擎、连接、表后生成 task.toml（非交互环境回退到内置样例；文件已存在则报错） |
db-ferry diff -task <name> | 对比指定任务的源库与目标库数据，支持 -keys、-where、-limit、-output、-format |
db-ferry mcp serve | 启动 MCP 服务器，提供 5 个 AI 原生工具 |
db-ferry -version | 查看版本号 |
db-ferry -sse-port :8080 | 启动 SSE 服务器，实时推送任务进度到 /events，状态查询 /status |
写入模式说明
| 模式 | 行为 | 风险 |
|---|
replace（默认） | DROP 目标表 → 重建 → 插入 | 目标表数据完全丢失 |
append | 直接 INSERT 追加数据 | 目标表已存在则报错（除非 skip_create_table=true） |
merge / upsert | 按 merge_keys 匹配，存在则更新，不存在则插入 | 目标表需有 merge_keys 对应的唯一约束 |
注意：upsert 是 merge 的别名，内部会自动统一为 merge。
增量续传配置
使用 resume_key + state_file 实现断点续传：
[[tasks]]
table_name = "orders"
sql = "SELECT order_id, order_date, total_amount FROM orders"
source_db = "prod"
target_db = "analysis"
mode = "append"
resume_key = "order_id"
state_file = "./state/orders.json"
validate = "row_count"
规则：
state_file 必须同时搭配 resume_keyresume_key 必须搭配 state_file 或 resume_from- SQL 中应保证
resume_key 字段单调递增
resume_from 为 SQL 字面量，表示从哪个值之后开始迁移（不包含该值）
常见错误与规避
| 错误 | 原因 | 解决 |
|---|
database definition: name is required | [[databases]] 缺少 name 字段 | 为每个数据库添加唯一 name |
duplicate database name | 数据库别名重复 | 修改为不同的 name |
source_db and target_db are both 'xxx'; set allow_same_table = true | 同库迁移未授权 | 添加 allow_same_table = true |
mode must be "replace", "append", "merge", or "upsert" | mode 值拼写错误 | 检查拼写，确认为四种之一 |
merge_keys is required when mode is "merge" | merge 模式缺少匹配键 | 添加 merge_keys = ["primary_key"] |
state_file requires resume_key | 配了 state_file 但没配 resume_key | 添加对应的 resume_key |
resume_key requires resume_from or state_file | 配了 resume_key 但没配 state_file 或 resume_from | 添加 state_file 或 resume_from |
partial indexes (where clause) are only supported for SQLite targets | 非 SQLite 目标使用了 where | 仅 SQLite 目标可使用部分索引 |
at least one database must be defined | 配置文件没有 [[databases]] 段 | 添加数据库定义 |
index name 'xxx' already defined | 索引名重复 | 修改为唯一的索引名 |
unsupported rule 'xxx' | 脱敏规则不存在 | 检查 rule 是否为内置 8 种之一 |
shard requires resume_key | 分片未配置 resume_key | 添加 resume_key 并确保 mode 为 append/merge |
adaptive_batch.min_size must be > 0 | 自适应批量参数缺失 | 补全 min_size、max_size、target_latency_ms、memory_limit_mb |
duplicate masking column / duplicate target column | masking 或 columns 中列重复 | 确保每列只出现一次 |
plugin.engine must be "lua" or "javascript" | 插件引擎不支持 | 检查 engine 是否为 lua 或 javascript |
plugin.script is required | 启用插件但未提供脚本 | 添加 script 字段 |
unsupported rule 'xxx' | 断言规则不存在 | 检查 rule 是否为内置 7 种之一 |
federated task requires at least 2 sources | 跨库 JOIN 源表不足 | 确保 sources 至少定义 2 个 |
join.keys is required for federated tasks | 跨库 JOIN 未指定连接键 | 添加 join.keys |
resume_key is not supported in federated mode | 联邦任务不支持断点续传 | 移除 resume_key 和 state_file |
invalid URL | notify 中的 Webhook URL 格式错误 | 检查 on_success/on_failure URL 格式 |
参考文件
详细的字段说明、验证规则和数据类型映射请参考：skills/db-ferry/references/config_reference.md
更多示例请参考：task.toml.sample、docs/user_guide.md
