ClientAdapter

基本说明

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管理接口, 运行目录结构为:

- bin
    restart.sh
    startup.bat
    startup.sh
    stop.sh
- lib
   ...
- plugin 
    client-adapter.logger-1.1.1-jar-with-dependencies.jar
    client-adapter.hbase-1.1.1-jar-with-dependencies.jar
    ...
- conf
    application.yml
    - hbase
        mytest_person2.yml
- logs

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

二、适配器配置介绍

2.1 总配置文件 application.yml

adapter定义配置部分

canal.conf:
  canalServerHost: 127.0.0.1:11111  # 对应单机模式下的canal server的ip:port
  zookeeperHosts: slave1:2181       # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准
  bootstrapServers: slave1:6667     # kafka或rocketMQ地址, 与canalServerHost不能并存
  flatMessage: true                 # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
  canalInstances:                   # canal实例组, 如果是tcp模式可配置此项
  - instance: example               # 对应canal destination
    groups:                         # 对应适配器分组, 分组间的适配器并行运行
    - outAdapters:                  # 适配器列表, 分组内的适配串行运行
      - name: logger                # 适配器SPI名
      - name: hbase
        properties:                 # HBase相关连接参数
          hbase.zookeeper.quorum: slave1
          hbase.zookeeper.property.clientPort: 2181
          zookeeper.znode.parent: /hbase
  mqTopics:                         # MQ topic租, 如果是kafka或者rockeMQ模式可配置此项, 与canalInstances不能并存
  - mqMode: kafka                   # MQ的模式: kafak/rocketMQ
    topic: example                  # MQ topic
    groups:                         # group组
    - groupId: g2                   # group id
      outAdapters:                  # 适配器列表, 以下配置和canalInstances中一样
      - name: logger                

说明:

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

adapter参数配置部分

adapter.conf:
  datasourceConfigs:                # 数据源配置列表, 数据源将在适配器中用于ETL、数据同步回查等使用
    defaultDS:                      # 数据源 dataSourceKey, 适配器中通过该值获取指定数据源
      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
      username: root
      password: 121212
  adapterConfigs:                   # 适配器内部配置列表
  - hbase/mytest_person2.yml        # 类型/配置文件名, 这里示例就是对应HBase适配器hbase目录下的mytest_person2.yml文件

三、适配器启动

3.1 启动canal-adapter示例

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

3.1.2 修改conf/application.yml为:

server:
  port: 8081
logging:
  level:
    com.alibaba.otter.canal.client.adapter.hbase: DEBUG
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null

canal.conf:
  canalServerHost: 127.0.0.1:11111
  flatMessage: true
  canalInstances:
  - instance: example
    adapterGroups:
    - outAdapters:
      - name: logger

3.1.3 启动

bin/startup.sh

3.2 adapter管理REST接口

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

curl http://127.0.0.1:8081/destinations

3.2.2 数据同步开关

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

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

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

3.2.3 数据同步开关状态

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

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

3.2.4 手动ETL

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

导入数据到指定类型的库

3.2.5 查看相关库总数据

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

适配器列表

logger适配器

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

Hbase适配器

同步HBase配置 : Sync-HBase

RDB适配器

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

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

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

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

ES适配器

MongoDB适配器

Redis适配器