基本说明

canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能:

  • 客户端启动器
  • 同步管理REST接口
  • 日志适配器, 作为DEMO
  • 关系型数据库的数据同步(表对表同步), ETL功能
  • HBase的数据同步(表对表同步), ETL功能
  • (后续支持) ElasticSearch多表数据同步,ETL功能

环境版本

  • 操作系统：无要求
  • java版本: jdk1.8 以上
  • canal 版本: 请下载最新的安装包，本文以当前v1.1.1 的canal.deployer-1.1.1.tar.gz为例
  • MySQL版本 ：5.7.18
  • HBase版本: Apache HBase 1.1.2, 若和服务端版本不一致可自行替换客户端HBase依赖

一、适配器整体结构

client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载, 目前所有支持的适配器都放置在plugin目录下

启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为:

1. - bin
2. restart.sh
3. startup.bat
4. startup.sh
5. stop.sh
6. - lib
7. ...
8. - plugin
9. client-adapter.logger-1.1.1-jar-with-dependencies.jar
10. client-adapter.hbase-1.1.1-jar-with-dependencies.jar
11. ...
12. - conf
13. application.yml
14. - hbase
15. mytest_person2.yml
16. - logs

以上目录结构最终会打包成 canal-adapter-*.tar.gz 压缩包

二、适配器配置介绍

2.1 总配置文件 application.yml

adapter定义配置部分

1. canal.conf:
2. canalServerHost: 127.0.0.1:11111          # 对应单机模式下的canal server的ip:port
3. zookeeperHosts: slave1:2181               # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准
4. mqServers: slave1:6667 #or rocketmq       # kafka或rocketMQ地址, 与canalServerHost不能并存
5. flatMessage: true                         # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
6. batchSize: 50                             # 每次获取数据的批大小, 单位为K
7. syncBatchSize: 1000                       # 每次同步的批数量
8. retries: 0                                # 重试次数, -1为无限重试
9. timeout:                                  # 同步超时时间, 单位毫秒
10. mode: tcp # kafka rocketMQ                # canal client的模式: tcp kafka rocketMQ
11. srcDataSources:                           # 源数据库
12. defaultDS:                              # 自定义名称
13. url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true   # jdbc url
14. username: root                                            # jdbc 账号
15. password: 121212                                          # jdbc 密码
16. canalAdapters:                            # 适配器列表
17. - instance: example                       # canal 实例名或者 MQ topic 名
18. groups:                                 # 分组列表
19. - groupId: g1                           # 分组id, 如果是MQ模式将用到该值
20. outerAdapters:                        # 分组内适配器列表
21. - name: logger                        # 日志打印适配器
22. ......

说明:

  1. 一份数据可以被多个group同时消费, 多个group之间会是一个并行执行, 一个group内部是一个串行执行多个outerAdapters, 比如例子中logger和hbase
  2. 目前client adapter数据订阅的方式支持两种，直连canal server 或者 订阅kafka/RocketMQ的消息

2.2 使用远程配置(Mysql)

可以使用远程配置中心(Mysql，可扩展)作为统一配置管理

2.1.1 创建mysql schema

1. CREATE SCHEMA `canal_manager` DEFAULT CHARACTER SET utf8mb4 ;

2.1.2 初始化数据

使用manager_ddl.sql脚本建表并初始化Demo数据，其中canal_config表id=2的数据对应adapter下的application.yml文件，canal_adapter_config表对应每个adapter的子配置文件

2.1.3 修改bootstrap.yml配置

1. canal:
2. manager:
3. jdbc:
4. url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8
5. username: root
6. password: 121212

可以将本地application.yml文件和其他子配置文件删除或清空， 启动工程将自动从远程加载配置

修改mysql中的配置信息后会自动刷新到本地动态加载相应的实例或者应用

三、适配器启动

3.1 启动canal-adapter示例

3.1.1 启动canal server (单机模式), 参考: Canal QuickStart

3.1.2 修改conf/application.yml为:

1. server:
2. port: 8081
3. logging:
4. level:
5. com.alibaba.otter.canal.client.adapter.hbase: DEBUG
6. spring:
7. jackson:
8. date-format: yyyy-MM-dd HH:mm:ss
9. time-zone: GMT+8
10. default-property-inclusion: non_null

12. canal.conf:
13. canalServerHost: 127.0.0.1:11111
14. batchSize: 500
15. syncBatchSize: 1000
16. retries: 0
17. timeout:
18. mode: tcp
19. canalAdapters:
20. - instance: example
21. groups:
22. - groupId: g1
23. outerAdapters:
24. - name: logger

3.1.3 启动

1. bin/startup.sh

3.2 adapter管理REST接口

3.2.1 查询所有订阅同步的canal instance或MQ topic

1. curl http://127.0.0.1:8081/destinations

3.2.2 数据同步开关

1. curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT

针对 example 这个canal instance/MQ topic 进行开关操作. off代表关闭, instance/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启

注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关

3.2.3 数据同步开关状态

1. curl http://127.0.0.1:8081/syncSwitch/example

查看指定 canal instance/MQ topic 的数据同步开关状态

3.2.4 手动ETL

1. curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST -d "params=2018-10-21 00:00:00"

导入数据到指定类型的库, 如果params参数为空则全表导入, 参数对应的查询条件在配置中的etlCondition指定

3.2.5 查看相关库总数据

1. curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml

适配器列表

logger适配器

1. 最简单的处理, 将受到的变更事件通过日志打印的方式进行输出, 如配置所示, 只需要定义name: logger即可
2. ...
3. outerAdapters:
4. - name: logger

Hbase适配器

同步HBase配置 : [[Sync-HBase]]

RDB适配器

同步关系型数据库配置 : [[Sync-RDB]]

目前内置支持的数据库列表:

  1. MySQL
  2. Oracle
  3. PostgresSQL
  4. SQLServer

使用了JDBC driver,理论上支持绝大部分的关系型数据库

ES适配器

同步关ES配置 : [[Sync-ES]]

MongoDB适配器

Redis适配器