molt-replicator

molt replicator

从源数据库持续捕获变更数据（CDC）并复制到 CockroachDB。在 molt fetch 完成初始批量加载之后运行。

重要：replicator 是独立于 molt 的单独二进制文件，不会由 molt fetch 调用。molt fetch 中的 data-load-and-replication 模式已弃用，请改为直接使用 replicator。

架构

Source DB ──► [replicator] ──► Staging DB (_replicator schema) ──► Target CockroachDB
                  ▲
            Publication/
            Slot/BinLog/
            LogMiner

Replicator 从源数据库读取变更，将其缓冲在目标 CRDB 集群的暂存架构中，然后应用到目标表。

按源数据库划分的子命令

源数据库	命令
PostgreSQL	replicator pglogical
MySQL	replicator mylogical
Oracle	replicator oraclelogminer
Kafka	replicator kafka
云存储	replicator objstore
CockroachDB CDC	replicator start

从 Fetch 到 Replicator 的完整工作流

第 1 步：使用 molt fetch 进行初始批量加载

drop-on-target-and-recreate 会先删除所有现有目标表；仅在全新或空目标上运行此命令。

molt fetch \
  --source "postgresql://<user>:<password>@source:5432/db" \
  --target "postgresql://root@crdb:26257/db" \
  --bucket-path "s3://mybucket/migration" \
  --table-handling drop-on-target-and-recreate

第 2 步：在源数据库上创建发布（PostgreSQL）

-- Run on source PostgreSQL:
CREATE PUBLICATION molt_fetch FOR ALL TABLES;
-- (molt fetch may have already created this; check first)

第 3 步：在目标数据库上创建暂存数据库

-- Run on target CockroachDB:
CREATE DATABASE _replicator;

第 4 步：测试连接

# preflight only takes --stagingConn and --targetConn (always required for the
# target; stagingConn required if the target is not CRDB)
replicator preflight \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --targetConn "postgresql://root@crdb:26257/db"

第 5 步：启动 replicator

replicator pglogical \
  --publicationName "molt_fetch" \
  --sourceConn "postgresql://<user>:<password>@source:5432/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public" \
  --metricsAddr "0.0.0.0:8080"

第 6 步：监控延迟

curl http://localhost:8080/metrics | grep replicator_
# Watch for: mutations applied, unapplied mutations, lag

第 7 步：切换

1. 当延迟接近 0 时，将应用写入重定向到 CockroachDB
2. 等待 replicator 处理完剩余变更
3. 确认源数据库上没有新的写入
4. 停止 replicator
5. 仅在 molt verify 通过且确认已备份源数据库后，才停用源数据库

特定源数据库的设置

PostgreSQL（pglogical）

源数据库前提条件：

• 用户具有 REPLICATION 权限
• 已启用逻辑复制（wal_level = logical）
• 发布已存在（由 molt fetch 创建或手动创建）

replicator pglogical \
  --publicationName "molt_fetch" \
  --slotName "replicator" \
  --sourceConn "postgresql://..." \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

MySQL（mylogical）

源数据库前提条件：

• 已启用二进制日志（binlog_format = ROW）
• 已启用 GTID 模式（gtid_mode=ON、enforce_gtid_consistency=ON）
• 用户具有 REPLICATION CLIENT 权限

replicator mylogical \
  --sourceConn "mysql://<user>:<password>@source:3306/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

Oracle（oraclelogminer）

源数据库前提条件：

• 已启用归档日志模式
• 已启用补充日志
• 已授予 LogMiner 权限

replicator oraclelogminer \
  --sourceConn "oracle://<user>:<password>@oracle:1521/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

关键标志

# Performance
--parallelism 16          # concurrent DB transactions (default: 16)
--flushSize 1000          # rows per batch (default: 1000)
--flushPeriod 1s          # flush interval (default: 1s)

# Staging connection pool
--stagingMaxPoolSize 128
--stagingIdleTime 1m
--stagingMaxLifetime 5m

# Target connection pool
--targetMaxPoolSize 128
--targetStatementCacheSize 128

# Retry
--maxRetries 10
--retryInitialBackoff 25ms
--retryMaxBackoff 2s

# Monitoring
--metricsAddr "0.0.0.0:8080"    # Prometheus metrics endpoint
--schemaRefresh 1m               # refresh schema cache (0 = disabled)

# Dead letter queue (failed rows instead of stopping)
--dlqTableName "replicator_dlq"

# Logging
-v                               # debug
-vv                              # trace
--logFormat fluent                # for log aggregators
--logDestination "/var/log/replicator.log"

注意事项

• 暂存架构（_replicator.public）由 replicator 自动创建，但必须先创建数据库（_replicator）
• --publicationName 和 --slotName 必须与 molt fetch 创建的名称一致。molt fetch 的 --pglogical-publication-name 默认值为 molt_fetch，而 --pglogical-replication-slot-name 没有默认值；在 replicator 端，--publicationName 没有默认值，--slotName 的默认值为 replicator。如果两端名称不一致，请在两端都显式设置。
• DLQ 表会随时间增长——请定期监控并清理失败的行
• Replicator 会在源数据库上持有一个打开的复制槽，这会阻止清理 WAL；请监控源数据库的磁盘使用量
• 优雅关闭会遵循 --gracePeriod（默认值：30s）；不要在未遵循此设置的情况下发送 SIGKILL
• 没有内置告警功能——请在 Prometheus 指标端点上设置外部告警
• 切换窗口过长会增加复制延迟——如有需要，请安排维护窗口

完整标志列表请参阅标志参考。