使用canal同步

使用流程

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

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

步骤一：部署Deployer

1. 下载canal.deployer包并解压。具体下载路径，请参见canal下载地址。
   解压后，您可以看到项目路径下的bin、conf、plugin、lib文件夹。
2. 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/路径下。
3. 修改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。

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

步骤二：部署Client-Adapter

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

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

   
3. 修改配置文件。
   1. 修改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**************************
      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

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

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

      1. 在conf/tablestore路径下创建.yml格式的文件，例如mytest.yml。
      2. 在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，则表示行更新，即一行数据更新时，只对变化的字段进行操作。

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

步骤三：同步MySQL全量数据

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

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

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

来源：https://help.aliyun.com/document_detail/307064.html