基于 MySQL + Tablestore 分层存储架构的大规模订单系统实践-数据同步 Canal 篇

前言 

在基于 MySQL + Tablestore 分层存储架构的大规模订单系统中，利用 CDC 技术将 MySQL 数据同步到 Tablestore 是不可缺少的一步。前文已经详细讲述了如何使用 DTS 向 Tablestore 同步数据。对于中小规模的数据库，或者个人开发者，还可以使用 Canal 从 MySQL 向 Tablestore 同步数据。Canal 部署简单，易于运维，且相比于 DTS，它成本更低，因此它更适合小规模的数据同步。

Canal 简介

Canal 是阿里开源 CDC 工具，他可以获取 MySQL binlog 解析，并将数据变动传输给下游。详情可参考Canal官网。基于 Canal，可以实现从 MySQL 到其他数据库的实时同步。Canal 部署简单、成本低，适合中小规模 Mysql 数据库同步工作。其架构如图：

Deployer 负责拉取 Binlog，解析数据，分发，记录位点。而 Client-adapter 负责接收上游数据，通过 Adapter适配器，将数据持久化到目标库。Deployer 和 Client-Adapter 作为 Canal 中的两个模块，分别独立部署。Tablestore 团队已经在 Adapter 中增加了 TablestoreAdapter，可以支持向 Tablestore 中写入数据。

下面，我们将部署 Canal，并将处于 Rds 中的 MySQL 订单数据同步进入 Tablestore，实现数据全量、增量的同步。

Canal 部署

环境准备

准备部署 canal 程序的机器。本文中在阿里云官网申请了一台 8 vCPU，16 GiB内存的 Linux 机器作为部署机器。如果读者同样需要申请 ECS，请参考：ECS入门概述。

源表准备

使用下面 SQL 在 MySQL 中新建测试表 order_contract_canal，其表结构与 order_contract 相同。

CREATE TABLE `order_contract_canal` (
  `oId` varchar(50) NOT NULL,
  `create_time` datetime NOT NULL COMMENT '下单时间',
  `pay_time` datetime DEFAULT NULL COMMENT '支付时间',
  `has_paid` tinyint(4) DEFAULT NULL COMMENT '是否已经支付',
  `c_id` varchar(20) DEFAULT NULL COMMENT '消费者id',
  `c_name` varchar(20) DEFAULT NULL COMMENT '消费者姓名',
  `p_brand` tinytext COMMENT '产品品牌',
  `p_count` mediumint(9) DEFAULT NULL COMMENT '产品数量',
  `p_id` varchar(20) DEFAULT NULL COMMENT '产品id',
  `p_name` varchar(20) DEFAULT NULL COMMENT '产品名',
  `p_price` decimal(16,2) DEFAULT NULL COMMENT '产品价格',
  `s_id` varchar(20) DEFAULT NULL COMMENT '店铺id',
  `s_name` varchar(20) DEFAULT NULL COMMENT '店铺名称',
  `total_price` decimal(16,2) DEFAULT NULL COMMENT '总价格',
  PRIMARY KEY (`oId`),
  KEY `idx_sid` (`s_id`),
  KEY `idx_paytime_sid` (`pay_time`,`s_id`) USING BTREE,
  KEY `idx_cid` (`c_id`),
  KEY `idx_paytime_cid_totalprice` (`pay_time`,`c_id`,`total_price`) USING BTREE,
  KEY `idx_sid_paytime` (`s_id`,`pay_time`),
  KEY `idx_sid_paytime_totalprice` (`s_id`,`pay_time`,`total_price`),
  KEY `idx_paytime_totalprice_pbrand` (`p_price`,`total_price`,`pay_time`) USING BTREE,
  KEY `idx_paytime` (`pay_time`),
  KEY `idx_pbrand_paytime` (`p_brand`(10),`pay_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

目标表准

在 Tablestore 中创建表 canal_target_order作为测试表，使其表结构与订单表 order_contract 相同。其表结构如图：

Deployer部署

在 Canal 官方 release页，下载 canal.deployer 包。将包解压后，配置 conf 路径下的配置文件，然后就可以通过 bin 路径下的启动脚本启动。配置 MySQL Binlog 模式以及部署 Canal Deployer 的具体步骤可以参考 Canal 官方文档的 QuickStart。

本文创建新的实例，在 Deployer 的 conf 目录下创建文件夹 test_ots。将 conf/example/instance.properties 复制到 test_ots 路径下。然后修改 test_ots 路径下的 instance.properties 配置文件。需要关注如下配置项：

ClientAdapter部署

在 Canal 官方 release页，下载 canal.adapter 包。将包解压后，配置 conf 路径下的配置文件，然后就可以通过 bin 路径下的启动脚本启动。包解压后，若 plugin 路径下不存在以 client-adapter.tablestore 开头的 jar 包，说明此安装包不包含 Tablestore 对接部分代码，需要使用 canal-adapter下的 zip 包作为安装包，该安装包与官网安装包部署路径相同，配置、启动方式相同。详细的部署步骤见 ClientAdapter。

conf 路径下 application.yml中需要额外关注配置见表。

完整配置 application.yml 配置如下

server:
  port: 8081
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null

canal.conf:
  mode: tcp #tcp kafka rocketMQ rabbitMQ
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: 3
  timeout:
  accessKey:
  secretKey:
  terminateOnException: true
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 127.0.0.1:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:

  srcDataSources:
    defaultDS:
      url: jdbc:mysql://rm-bp15p07134rkvf7z6wo.mysql.rds.aliyuncs.com:3306/test_ots?useUnicode=true
      username: ****
      password: ****
  canalAdapters:
  - instance: test_ots # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: tablestore
        key: ts
        properties:
          tablestore.endpoint: https://test-20210609.cn-hangzhou.ots.aliyuncs.com
          tablestore.accessSecretId: ****
          tablestore.accessSecretKey: ****
          tablestore.instanceName: test-20210609

在 conf/tablestore 路径下，创建 order.yml 文件，填入以下内容。配置表示从源库表 test_ots. order_contract_canal 向目标表 canal_target_order 同步数据。

dataSourceKey: defaultDS
destination: test_ots
groupId: g1
outerAdapterKey: ts
threads: 8
updateChangeColumns: false
dbMapping:
  database: test_ots
  table: order_contract_canal
  targetTable: canal_target_order
  targetPk:
    oId: oId
  targetColumns:
    oId:
    create_time:
    pay_time:
    has_paid:
    c_id:
    c_name:
    p_brand:
    p_count:
    p_id:
    p_name:
    p_price:
    s_id:
    s_name:
    total_price:
  etlCondition: 
  commitBatch: 200 # 批量提交的大小

其中各参数含义见表。

全量同步数据

使用程序向原始表 order_contract_canal 中插入 1 千万行记录。

同步数据

调用 client-adapter 服务的方法触发同步任务。指令格式为

curl "localhost:8081/etl/{type}/{key}/{task}" -X POST

type 为下游数据库类型；key 是 adapter key；task 为任务配置文件的名称。在本文中，指令为：

curl "localhost:8081/etl/tablestore/ts/order.yml" -X POST

程序会首先中止增量数据传输，然后同步全量历史数据。同步开始后，可以在日志中看到 Adapter 中 TablestoreWriter 的传输日志变化。

性能测试

在 Tablestore 监控页面查看数据写入速率，首先进入 Tablestore控制台。点击对应实例进入实例详情页。

点击表 canal_target_order。

点击监控指标进入监控页面。

同步任务开始后，在监控页面可以看到数据如图。此时 Canal 所在机器配置为 8 核 16G，order.yml 中 threads 配置为 8。源库记录数在 1千万，每行数据大小约 0.5KB。可以看到在任务开始的时候，并发写入 Tablestore 速率很高，在 2w行/s 左右。而随着任务的进行，写入速率开始下降，这是由于全量导入数据时从源库获取数据时使用 limit offset 导致的，受限于上游数据的获取。

1千万数据完成写入共耗时 28 分钟。耗时统计见下表。可以看到，在数据导入初期，导入速率相对较快，而数据导入后期，导入效率明显降低。时间统计和控制台中监控数据吻合。

增量同步数据

使用附录中的程序中的接口（"/canal/press"），向原始表持续写入数据。

性能测试

输入如下指令调用接口，使用 3 线程写入数据，每个线程每秒写入 4000 行记录。

curl "localhost:8082/canal/press?rps=4000&threads=3"  -X POST

在控制台可以看到数据持续写入，速率约在 1.2w 行/s。

在当前 8 核 16G 的机器配置下，继续增加并发写入量，写入 1.6w 到 2w 行每秒，测试出的增量同步上限约在 1.5w 行/s，每行记录约 0.5KB。

异常处理

ClientAdapter 配置文件application.yml 中 terminateOnException 若不配置或配置为 false，同步程序同步后仍报错，则程序会记录日志，跳过报错数据，继续同步任务。而若 terminateOnException 配置为 true，则同步报错后，程序会中止增量数据同步任务，等待用户介入处理报错。此时，用户可以通过下面接口查看任务的开启、中断状态。命令格式如下：

curl "localhost:8081/syncSwitch/{destination}"

在本文中命令为

curl "localhost:8081/syncSwitch/test_ots"

处理异常后，可以调用如下接口重新启动增量同步任务。

curl "localhost:8081/syncSwitch/test_ots/on" -X PUT

总结

本文简要介绍了 Canal，并且详细的展示了如何使用 Canal 从 MySQL 库向 Tablestore 中同步全量、增量数据。

附录

Canal 测试程序git地址：

https://github.com/aliyun/tablestore-examples