10-biz-mysql-table-sharding-cdc

2.1 先否定一个直觉方案#

最直观的做法：开启 Canal 订阅增量 Binlog，同时用 Canal RDB Adapter 内置的 ETL 功能全量拉取主表存量数据，最后改应用配置完成切表。

看似闭环，放到生产环境有四个致命隐患：

2.2 五步法#

业界标准做法可以概括为「双写 + 追平 + 校验 + 反向同步」，拆成五步：

第一步：增量订阅与位点标记（Prepare）。启动 Canal Deployer 订阅旧主表 Binlog，记录启动瞬间的 GTID 或 Binlog File + Position。增量 Binlog 可以挂载到 Kafka 等消息队列，也可以直接推给 Canal Adapter。增量链路先跑起来，且所有写入分表的逻辑必须幂等：INSERT INTO ... ON DUPLICATE KEY UPDATE，或基于 update_time 时间戳做版本判定，防止旧数据覆盖新数据。启动前还有一项 DDL 预检：源表若有承载状态值的 tinyint(1) 列，先改成 tinyint(4)——这类列在 Canal 解析阶段就会被折叠成 boolean，事后任何下游配置都救不回来（见第六节）。

第二步：存量数据迁移（Full Migration）。全量抽取严禁直接在主库执行，选择从库（Read Replica）或备份节点。工具用 DataX、mydumper 这类成熟的离线同步方案，或按主键 ID 做 Chunk 切片分段抽取，并限制抽取速率保护数据库。存量写入分表遇到主键冲突时用 INSERT IGNORE，或仅在存量数据时间戳更新时才执行更新，确保已被增量 Binlog 写入的新数据不被覆盖。

第三步：增量追平与数据校验（Validation）。存量导入完成后，让分表继续消费积压的 Binlog，直到延迟降至毫秒级。然后是校验，三个层次：

• 静态 Count & Sum：比对主表与分表在特定时间范围内的总条数，以及关键数值字段的和；
• 采样与全量 Hash：对主键取模采样，比对两边行记录的 Hash 值，条件允许就跑全量；
• 差异修复：运行独立的对比修复程序，扫描差异并自动生成补偿，直到一致率 100%。

第四步：灰度切流与反向同步（Cutover）。切流之前先搭好反向同步通道：用 Canal 订阅分表 Binlog，实时写回主表，让系统随时具备一键回滚的能力。切流本身分三步推进：

1. 影子读（Shadow Read）：应用部署新代码，写流量仍走主表，开启双读，对比主表与分表的读取结果，打日志监控差异；
2. 灰度写：按用户 ID、租户 ID 等路由键，把写流量逐步切到分表，1% → 10% → 50% → 100%；
3. 完全切流：所有读写彻底切到分表，观察 24~48 小时。发现异常，直接利用反向同步补齐的主表数据恢复流量。

第五步：链路收尾（Cleanup）。分表稳定运行一周、确认不再需要回滚之后：停止 Canal 正向与反向同步服务，下线主表同步链路相关组件，旧主表归档并释放数据库资源。

2.3 并发冲突在 SQL 层面根治#

存量抽取与增量消费并行，旧数据可能后到。根治办法是让写入 SQL 自带版本判定：

1
-- 增量 Binlog 写入分表：幂等 + update_time 守卫，状态只前进不后退
2
INSERT INTO logic_1 (id, base_id, f_key, state, updated_at)
3
VALUES (?, ?, ?, ?, ?)
4
ON DUPLICATE KEY UPDATE
5
  state = IF(VALUES(updated_at) >= updated_at, VALUES(state), state),
6
  updated_at = IF(VALUES(updated_at) >= updated_at, VALUES(updated_at), updated_at);
7

8
-- 存量数据写入分表：INSERT IGNORE，给增量让路
9
INSERT IGNORE INTO logic_1 (id, base_id, f_key, state, updated_at)
10
VALUES (?, ?, ?, ?, ?);

增量写入用 updated_at 做守卫；存量写入直接放弃冲突行。反向回写链路复用同一套幂等 SQL，还有一个附带的好处：MySQL ROW 模式的 Binlog 不会为「没有改变任何列值」的更新产生新事件，正向与反向链路即使互相触发，回环也会因为数据收敛而自然停止，不会无限放大。

2.4 回滚预案是方案的一部分，不是保险#

迁移方案的成败不只取决于上线是否顺畅，更取决于出问题时能否优雅退回。反向同步链路必须在灰度切流之前完全就绪。一旦分表出现未知死锁、慢查询拖垮业务这类紧急情况，把应用开关切回主表，业务分钟级恢复。

3.1 为什么需要聚合#

分表落地后，新的痛点跟着来：物理分表散落，原生跨表 UNION ALL 聚合查询性能极差；前端统计分析高频，不能挤占线上主库的 CPU 与 IO；业务要求数据变更后亚秒级反映到统计面板，物理删除也要实时同步。

方案是把 12 张分表实时汇聚到同一个 ES 索引 logic_all_index，查询压力全部甩给 ES，与主库彻底读写解耦。

3.2 架构与数据流转#

结合运维成本与扩展性，采用 Canal Deployer + Canal Adapter 的标准轻量组合。若后续写并发突破千级 QPS，可在中间引入 Kafka/RocketMQ 削峰填谷。

数据流转四步：

1. 捕获（Dump）：Deployer 伪装成 MySQL Slave 向 Master 发送 dump 协议，读取原始 Binlog；
2. 解析（Parse）：二进制日志解析为结构化的 FlatMessage，包含变更前后的列名与列值；
3. 映射（Transform）：Adapter 匹配正则 logic_.*，按 YAML 规则把 MySQL 字段转为 ES Document；
4. 加载（Load）：触发 Bulk 攒批机制，调用 ES REST API 批量写入。

3.3 先显式建 Mapping#

开启同步前必须在 ES 中显式创建 Mapping。f_key 是 varchar，动态 Mapping 会把它识别成 text，Terms Aggregation 直接报 Fielddata is disabled on text fields。日期字段同时声明三种格式兼容，避免源端时间格式差异导致解析失败：

1
PUT /logic_all_index
2
{
3
  "mappings": {
4
    "properties": {
5
      "id": { "type": "long" },
6
      "base_id": { "type": "long" },
7
      "f_key": { "type": "keyword", "doc_values": true },
8
      "state": { "type": "integer" },
9
      "created_at": {
10
        "type": "date",
11
        "format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
12
      },
13
      "updated_at": {
14
        "type": "date",
15
        "format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
16
      }
17
    }
18
  }
19
}

3.4 Adapter 映射配置#

在 Adapter 的 conf/es7/ 目录下添加映射配置 logic_table_es.yml：

1
dataSourceKey: defaultDS # 对应 application.yml 中的 MySQL 数据源配置
2
destination: example # 对应 Canal Deployer 的 instance 名称
3
groupId: g1 # 消费分组 ID
4
outerAdapterKey: es7_key # 对应全局配置中 ES 适配器的 Key
5
concurrent: true # 开启并发同步，提升吞吐量
6

7
dbMapping:
8
  database: my_biz_db # 数据库名称
9
  table: logic_.* # 【核心】正则匹配 logic_1 到 logic_12
10
  index: logic_all_index # 目标 ES 索引
11
  type: _doc # ES 7.x/8.x 统一填 _doc
12
  pk: id # 物理分表的主键列
13
  id: f_key # 【核心】ES 文档 _id：全局唯一业务键
14
  upsert: true # 存在则更新，不存在则插入
15
  commitBatch: 1000 # 达到 1000 条触发一次 Bulk 提交
16
  commitInterval: 1000 # 或每隔 1000ms 强制触发一次 Bulk 提交
17
  fields: # 字段映射字典（DB_Column : ES_Field）
18
    id: id
19
    base_id: base_id
20
    f_key: f_key
21
    state: state
22
    created_at: created_at
23
    updated_at: updated_at

几个关键决策：

• table: logic_.* 一条正则覆盖 12 张物理分表，后续扩表也不需要改同步配置；
• id: f_key 指定 ES 文档 _id。f_key 具备全局唯一性可以直接用；如果没有全局唯一业务键，必须用 CONCAT(base_id, '_', id) 之类的方式合成。12 张分表各自使用表内自增 ID，logic_1 和 logic_2 完全可能都有 id=100，直接拿 id 当 _id 会造成跨表数据互相覆盖，这是高危项；
• upsert: true 幂等覆写，配合一致的 _id 是全量与增量对接的基础（见 3.5）；
• commitBatch / commitInterval 双阈值攒批，先到先触发，避免单条写入打垮 ES。

3.5 存量 + 增量无缝对接#

Canal 作为 CDC 工具，默认读不到服务启动前已经存在的历史数据，全量需要单独走一条路。按数据量选：

• Adapter 内置 ETL（百万级以内）：部署完 Adapter 后直接调 REST 接口触发全量拉取，Adapter 自动扫描 12 张物理表并发抽取并 Bulk 写入 ES：

  1
  curl -X POST http://127.0.0.1:8081/etl/es7/logic_table_es.yml

• DataX / 离线脚本（千万级以上）：用 DataX 的 mysqlreader 多线程并发抽取 12 张表，elasticsearchwriter 批量写入 ES。

顺序上的铁律：先启动增量链路，再触发全量抽取。全量跑的这段时间里，线上业务还在持续产生 Binlog，必须防止全量的旧数据覆盖增量刚写入的新数据。两种机制：

• 机制 A：Upsert 覆写（首选，简单高效）。全量与增量使用完全一致的 ES _id（如 f_key），写入行为都是 Upsert。即使旧的全量数据后到达 ES，最新的增量 Binlog 变更也会随后再次覆写该 _id，保证最终一致性；
• 机制 B：外部版本控制（External Versioning）。ES 写入时指定 version_type=external，把 MySQL 的 updated_at 时间戳作为 version。ES 引擎拒绝任何版本号小于等于现有文档的写请求，物理级阻断「旧盖新」。

3.6 TINYINT 在 ES 侧的表现#

tinyint(1) 之外的 TINYINT 经 Canal 解析后是数字，到了 ES 侧最终表现为 Integer 还是 Boolean，完全取决于 Mapping 定义和 ES 的类型强转（Coercion）机制：

• Mapping 设为 "state": { "type": "integer" }，写入 1, 2, 3 正常存储；
• Mapping 设为 "is_deleted": { "type": "boolean" }，按 ES 默认 Coercion 规则，1、"1"、"true" 一律强转为 true，0、"0"、"false" 强转为 false。

tinyint(1) 是例外：它在 Canal 解析侧就被折叠成 boolean，原始状态值在到达 ES 之前已经丢失（实测案例见第六节），Mapping 定义得再对也救不回来。

如果需要把 TINYINT 状态码（1, 2）转义成可读文本（PENDING, SUCCESS），可以在 Adapter YAML 中编写 sql: 配置项，调用 MySQL 的 CASE WHEN 在读取阶段完成清洗。但要注意代价：配置了 sql: 的映射会让 Adapter 在增量同步时反查源库，写入量大的表会明显拖慢链路（见第六节）。

4.1 一源双送#

利用 Canal 的一源多送特性，Deployer 只解析一份 Binlog 镜像，Adapter 层同时驱动 ES Adapter 与 RDB (MySQL) Adapter：

切流期间，这条反向链路的回写目标是旧主表，充当回滚兜底；迁移稳定后，回写目标可以是独立的物理总表 logic_total，方便第三方 BI 报表工具直连或做逻辑归档。两种用途机制完全相同，下面以 logic_total 为例。

4.2 全局连接配置#

conf/application.yml 中同时挂载 ES 与 MySQL RDB 两个 Outer Adapter。重点在 MySQL 的 JDBC URL，必须追加类型对齐参数（原因见第五节）：

1
canalAdapters:
2
  - instance: example # Canal Deployer 实例名称
3
    groups:
4
      - groupId: g1
5
        outerAdapters:
6
          # --- 目标端 1：Elasticsearch ---
7
          - key: es7_key
8
            name: es7
9
            hosts: 192.168.1.100:9200
10

11
          # --- 目标端 2：MySQL 物理总表 ---
12
          - key: mysql_total_key
13
            name: rdb
14
            properties:
15
              jdbc.driverClassName: com.mysql.cj.jdbc.Driver
16
              # 【核心】必加 tinyInt1isBit=false&serverTimezone=Asia/Shanghai
17
              jdbc.url: jdbc:mysql://192.168.1.200:3306/total_db?useSSL=false&charset=utf8mb4&tinyInt1isBit=false&transformedBitIsBoolean=false&zeroDateTimeBehavior=CONVERT_TO_NULL&serverTimezone=Asia/Shanghai
18
              jdbc.username: root
19
              jdbc.password: 123456

4.3 RDB Adapter 映射配置#

conf/rdb/logic_table_mysql.yml：

1
dataSourceKey: defaultDS
2
destination: example
3
groupId: g1
4
outerAdapterKey: mysql_total_key
5
concurrent: true
6

7
dbMapping:
8
  database: my_biz_db
9
  table: logic_.* # 匹配源端 12 张分表
10
  targetDb: total_db
11
  targetTable: logic_total # 写入目标物理总表
12
  targetPk:
13
    f_key: f_key # 以 f_key 唯一键防重
14
  mapAll: true # 自动映射字段
15
  targetColumns:
16
    id: id
17
    base_id: base_id
18
    f_key: f_key
19
    state: state
20
    updated_at: updated_at

targetPk 用 f_key 防重，mapAll: true 自动映射同名字段，targetColumns 显式对齐关键列。

4.4 落地 SOP#

1. 提前建表：创建 ES Index Mapping（f_key 指定为 keyword）与 MySQL 总表 logic_total（建立唯一主键）；
2. 挂载增量：启动 Canal Deployer 与 Adapter，让增量同步链路先跑起来；
3. 触发全量：调用 Adapter 的 ETL 接口，或运行 DataX/SeaTunnel 脚本抽取历史数据；
4. 幂等对齐：依赖 ES / MySQL 的 Upsert 机制自动冲刷旧数据，保证最终一致性。

4.5 两条致命红线#

1. 总表主键冲突。12 张分表若存在自增 ID 冲突，写入总表必须使用 f_key 或 (base_id, id) 联合主键。严禁直接拿分表自增 id 作总表唯一主键，否则跨表数据互相覆盖。

2. 无限循环同步。若总表与源分表放在同一个 MySQL 实例中，必须在 Canal Deployer 的 instance.properties 里用正则严格圈定订阅范围、排除总表：

   1
   canal.instance.filter.regex = my_biz_db\\.logic_([1-9]|1[0-2])

   否则回写总表的动作产生新 Binlog，又被 Canal 捕获再回写，形成死循环。