如何使用canal将MySQL数据同步到表格存储_表格存储(Tablestore)-阿里云帮助中心

背景信息

canal基于MySQL数据库增量日志解析，提供增量数据订阅和消费，是阿里开源CDC工具，它可以获取MySQL binlog数据并解析，然后将数据变动传输给下游。基于canal，您可以实现从MySQL到其他数据库的实时同步。更多信息，请参见canal官网。

使用流程

使用canal将MySQL数据同步到表格存储的使用流程如下：

部署Deployer服务，该服务负责从上游拉取binlog数据、记录位点等。具体操作，请参见步骤一：部署Deployer。
部署Client-Adapter服务，该服务负责对接Deployer解析过的数据，并将数据传输到目标库中。具体操作，请参见步骤二：部署Client-Adapter。
部署完成后，canal默认会自动同步MySQL增量数据。
如果需要同步MySQL全量数据，请手动调用Client-Adapter服务的方法触发同步任务。具体操作，请参见步骤三：同步MySQL全量数据。
待全量数据同步完成后，canal会自动开始增量同步。

步骤一：部署Deployer

说明由于Deployer包中不存在为表格存储定制的jar包或者配置，因此使用canal对接表格存储时的部署方式与使用canal对接MySQL时的部署方式类似。具体操作，请参见canal快速入门。

下载canal.deployer包并解压。具体下载路径，请参见canal下载地址。
解压后，您可以看到项目路径下的bin、conf、plugin、lib文件夹。
canal实例名称默认为example，如果需要自定义canal实例名称，例如改为test_ots，请执行以下操作。
1. 修改canal.properties文件中canal.destinations的值为自定义的canal实例名称，其他配置均保持默认即可。
```
canal.destinations = test_ots  
```
2. 在conf路径下创建以canal实例名称命名的文件夹test_ots，并将conf/example路径下的instance.properties文件复制到conf/test_ots/路径下。

修改instance.properties文件。instance.properties文件中的主要配置项说明请参见下表。


配置项	是否必填	示例值	描述
canal.instance.master.address	是	rm-bp15p07134rkvf7**.mysql.rds.aliyuncs.com:**	canal监听的数据库地址，格式为`host:port`。
canal.instance.rds.accesskey	否	LTAn*********************	当MySQL为阿里云产品RDS库时，填写登录账号的AccessKey ID和AccessKey Secret，获取方式请参见为RAM用户创建访问密钥。如果非RDS库，无需填写此项。
canal.instance.rds.secretkey	否	zbnK**************************
canal.instance.rds.instanceId	否	rm-bp15p0713****	当MySQL为阿里云产品RDS库时，填写实例ID。如果非RDS库，无需填写此项。
canal.instance.dbUsername	是	test	数据库账号用户名。
canal.instance.dbPassword	是	db****	数据库账号密码。
canal.instance.filter.regex	是	.\\..	canal实例关注的表。通过正则表达式匹配。此处表示匹配所有数据库下的所有表。
canal.destinations	是	test_ots	canal实例名称，必须与配置文件所在上层路径相同。例如配置文件的路径为conf/test_ots/instance.properties，则canal实例名称为test_ots。

通过bin路径下的脚本启动项目。

步骤二：部署Client-Adapter

下载canal.adapter包并解压。具体下载路径，请参见canal下载地址。
解压后，您可以看到项目路径下的bin、conf、plugin、lib文件夹。
如果plugin路径下不存在以client-adapter.tablestore开头的jar包，请下载canal-adapter部署包并解压，然后使用该部署包的内容替代原有的canal.adapter包。

说明如果plugin路径下存在以client-adapter.tablestore开头的jar包，则说明部署包中已包含canal对接表格存储的适配器代码，此时可以直接使用该部署包。

修改配置文件。

修改conf路径下application.yml文件中表格存储相关的配置信息，详细配置项说明请参见下表，其他配置项的说明请参见ClientAdapter配置项说明。


配置项	是否必填	示例值	描述
canal.conf:canalAdapters：instance	是	test_ots	canal实例名称，必须与部署Depolyer时自定义的canal实例名称相同。
canal.conf:canalAdapters：outerAdapters: -name:	是	tablestore	定义适配器类型。设置此项为tablestore，表示此适配器下游写入Tablestore库。
canal.conf:canalAdapters：outerAdapters: properties：tablestore.endpoint	是	https://test-2009.cn-hangzhou.ots.aliyuncs.com	填写目标Tablestore实例的服务地址。
canal.conf:canalAdapters：outerAdapters: properties：tablestore.accessSecretId	是	LTAn********************	登录账号的AccessKey ID和AccessKey Secret，获取方式请参见为RAM用户创建访问密钥。
canal.conf:canalAdapters：outerAdapters: properties：tablestore.accessSecretKey	是	zbnK**************************	登录账号的AccessKey ID和AccessKey Secret，获取方式请参见为RAM用户创建访问密钥。
canal.conf:canalAdapters：outerAdapters: properties：tablestore.instanceName	是	test-2009	Tablestore实例的名称。
canal.conf: syncBatchSize	否	1000	一个请求批次中涉及的行数，统计DML（Data Manipulation Language）操作中的数据大小。如果一次拉取到的DML中涉及到的行数过大，且行数大于此参数值，则该批次数据会被拆分处理。
canal.conf: retries	否	3	Adapterprocessor层的重试次数。默认值为3。
canal.conf: timeout	否	1000	Adapterprocessor层通过consumer向上游拉取数据的超时时间。单位为毫秒。如果设置timeout为0，则拉取数据时，Adapterprocessor层会处于阻塞状态直到拉取到数据。
canal.conf : consumerProperties:canal.tcp.batch.size	否	500	consumer向上游拉取的DML个数。
canal.conf: terminateOnException	否	true	默认为false。如果配置为true，则数据同步重试后仍失败，程序会暂停实时同步任务，等待用户手动处理。

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-bp15po.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-2009.cn-hangzhou.ots.aliyuncs.com
          tablestore.accessSecretId: ****
          tablestore.accessSecretKey: ****
          tablestore.instanceName: test-2009

在conf/tablestore路径下创建.yml格式的文件并配置。

说明如果不存在conf/tablestore路径，请手动创建该路径。

在conf/tablestore路径下创建.yml格式的文件，例如mytest.yml。

在mytest.yml文件中填写以下内容并配置相应参数。

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: createTime$string
    pay_time: $string
    update_time: updateTime
  etlCondition: 
  commitBatch: 200 # 批量提交的行数。


配置项	是否必填	示例值	描述
dataSourceKey	是	defaultDS	该任务的源数据库标识，您可以在application.yml中canal.conf: srcDataSources下找到该标识对应的数据库。
destination	是	test_ots	canal实例名称，必须与application.yml中的canal.conf: canalAdapters下instance值相同。
groupId	是	g1	分组ID，与application.yml中canal.conf: canalAdapters: groups下groupId值相同即可。在MQ模式下才需要使用，此处无需关注。
outerAdapterKey	是	ts	使用的Adapter标识，必须与application.yml中canal.conf: canalAdapters: groups: outerAdapters下key值相同。
threads	是	8	Bucket数量，默认值为1，对应于tablestorewriter中的bucket数量。
dbMapping.database	是	test_ots	源数据库名称。
dbMapping.table	是	order_contract_canal	源表名称。
dbMapping.targetTable	是	canal_target_order	目标表名称。
dbMapping.targetPk	是	oId: oId	主键配置，格式为`id: target_id`，即`源表主键: 目标表主键`。当表存在多个主键时需要配置多个，多个主键配置顺序必须与Tablestore中的主键顺序相同。 dbMapping.targetPk中支持配置主键列自增，格式为`$$: target_id`，表示在目标表中生成一列名称为target_id的主键且该主键列为自增列。当上游数据写入Tablestore时，canal adapter会自动填充该列的值。关于主键列自增的更多信息，请参见主键列自增。
dbMapping.targetColumns	是	create_time: createTime$string	配置需要同步的列名以及列映射，支持配置类型转换。注意在dbMapping.targetPk中配置的非自增的主键列也需要在此处再进行配置，自增主键列不需要再进行配置。如果不配置类型转换，则canal会根据源表中字段类型推断目标字段类型。更多信息，请参见源表和目标Tablestore表中字段映射。支持配置的格式包含如下4种，请根据实际需要配置。说明在配置映射的字段类型时，字段类型大小写不敏感。 `id: target_id$string`：表示源表中id字段同步到目标表后为target_id字段，且字段类型映射为string。 `id: target_id`：表示源表中id字段同步到目标表后为target_id字段。 `id:`：表示id字段同步前后字段名不变，字段类型采用默认映射。 `id: $string`：等同于`id: id$string`，表示id字段同步前后字段名不变，且字段类型映射为string。
dbMapping.etlCondition	否	where create_time > "2021-01-01"	全量抽取数据时的过滤条件，其中字段名称为源表字段名称。
dbMapping.commitBatch	否	200	一次批量RPC请求导入的行数，对应于tablestorewriter中的maxBatchRowsCount，默认为writerConfig中的默认值200。
updateChangeColumns	否	false	行覆盖或行更新。默认值为false，表示行覆盖，即一行数据更新时，使用该行最新的整行值覆盖Tablestore中的旧行。如果设置为true，则表示行更新，即一行数据更新时，只对变化的字段进行操作。

通过bin路径下的脚本启动项目。

步骤三：同步MySQL全量数据

执行以下命令调用Client-Adapter服务的方法触发同步任务。此时，canal会先中止增量数据传输，然后同步全量数据。待全量数据同步完成后，canal会自动进行增量数据同步。

命令格式

curl "hostip:port/etl/type/key/task" -X POST

示例

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

详细配置项说明请参见下表。


配置项	是否必选	示例	描述
hostip	是	localhost	部署canal服务的机器IP地址和端口。当在部署canal服务的机器上执行此命令时，可设置hostip为localhost。
port	是	8081
type	是	tablestore	下游数据库类型，必须设置为tablestore。
key	是	ts	使用的Adapter标识，必须与application.yml中canal.conf: canalAdapters: groups: outerAdapters下key值相同。
task	是	mytest.yml	任务配置文件的名称，必须与步骤二：部署Client-Adapter中创建的.yml格式的文件名称相同。

全量数据同步开始后，您可以在日志中查看Adapter中TablestoreWriter的传输日志变化，如下图所示。 fig_datasyncmysqlall

源表和目标Tablestore表中字段映射

canal支持源表到目标Tablestore表的字段名称以及字段类型映射，对应于dbMapping.targetColumns字段中的映射配置。支持作为目标类型配置在$后面的字段类型请参见下表。


源表中字段类型	目标Tablestore表中字段类型
string	string
int	int
integer	int
bool	bool
boolean	bool
binary	binary
double	double
float
decimal

使用canal同步

前提条件