基本说明

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

  • 客户端启动器
  • 同步管理REST接口
  • 日志适配器, 作为DEMO
  • 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的方式让启动器动态加载

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

  1. canal-adapter-launcher.jar
  2. - lib
  3.     client-adapter.logger-1.1.1-jar-with-dependencies.jar
  4.     client-adapter.hbase-1.1.1-jar-with-dependencies.jar
  5. - config
  6.     application.yml
  7.     - hbase
  8.         mytest_person2.yml

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

二、启动器

2.1 启动器配置 application.yml

canal相关配置部分说明

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

适配器相关配置部分说明

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

2.2 同步管理REST接口

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

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

2.2.2 数据同步开关

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

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

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

2.2.3 数据同步开关状态

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

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

2.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"

导入数据到指定类型的库

2.2.5 查看相关库总数据

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

2.3 启动canal-adapter示例

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

2.3.2 修改config/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.   flatMessage: true
  15.   canalInstances:
  16.   - instance: example
  17.     adapterGroups:
  18.     - outAdapters:
  19.       - name: logger

启动 canal-adapter-launcher.jar

  1. java -jar canal-adapter-launcher.jar

三、HBase适配器

3.1 修改启动器配置: 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.   flatMessage: true
  15.   canalInstances:
  16.   - instance: example
  17.     adapterGroups:
  18.     - outAdapters:
  19.       - name: hbase
  20.         properties:
  21.           hbase.zookeeper.quorum: slave1
  22.           hbase.zookeeper.property.clientPort: 2181
  23.           zookeeper.znode.parent: /hbase
  25. adapter.conf:
  26.   datasourceConfigs:
  27.     defaultDS:
  28.       url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
  29.       username: root
  30.       password: 121212
  31.   adapterConfigs:
  32.   - hbase/mytest_person.yml

其中指定了一个HBase表映射文件: mytest_person.yml

3.2 适配器表映射文件

修改 config/hbase/mytest_person.yml文件:

  1. dataSourceKey: defaultDS            # 对应application.yml中的datasourceConfigs下的配置
  2. hbaseOrm:                           # mysql--HBase的单表映射配置
  3.   mode: STRING                      # HBase中的存储类型, 默认统一存为String, 可选: #PHOENIX  #NATIVE   #STRING 
  4.                                     # NATIVE: 以java类型为主, PHOENIX: 将类型转换为Phoenix对应的类型
  5.   destination: example              # 对应 canal destination/MQ topic 名称
  6.   database: mytest                  # 数据库名/schema名
  7.   table: person                     # 表名
  8.   hbaseTable: MYTEST.PERSON         # HBase表名
  9.   family: CF                        # 默认统一Column Family名称
  10.   uppercaseQualifier: true          # 字段名转大写, 默认为true
  11.   commitBatch: 3000                 # 批量提交的大小, ETL中用到
  12.   #rowKey: id,type                  # 复合字段rowKey不能和columns中的rowKey并存
  13.                                     # 复合rowKey会以 '|' 分隔
  14.   columns:                          # 字段映射, 如果不配置将自动映射所有字段, 
  15.                                     # 并取第一个字段为rowKey, HBase字段名以mysql字段名为主
  16.     id: ROWKE                       
  17.     name: CF:NAME
  18.     email: EMAIL                    # 如果column family为默认CF, 则可以省略
  19.     type:                           # 如果HBase字段和mysql字段名一致, 则可以省略
  20.     c_time: 
  21.     birthday:

如果涉及到类型转换,可以如下形式:

  1. ...
  2.   columns:                         
  3.     id: ROWKE$STRING                      
  4.     ...                   
  5.     type: TYPE$BYTE                          
  6.     ...

类型转换涉及到Java类型和Phoenix类型两种, 分别定义如下:

  1. #Java 类型转换, 对应配置 mode: NATIVE
  2. $DEFAULT
  3. $STRING
  4. $INTEGER
  5. $LONG
  6. $SHORT
  7. $BOOLEAN
  8. $FLOAT
  9. $DOUBLE
  10. $BIGDECIMAL
  11. $DATE
  12. $BYTE
  13. $BYTES
  1. #Phoenix 类型转换, 对应配置 mode: PHOENIX
  2. $DEFAULT                  对应PHOENIX里的VARCHAR
  3. $UNSIGNED_INT             对应PHOENIX里的UNSIGNED_INT           4字节
  4. $UNSIGNED_LONG            对应PHOENIX里的UNSIGNED_LONG          8字节
  5. $UNSIGNED_TINYINT         对应PHOENIX里的UNSIGNED_TINYINT       1字节
  6. $UNSIGNED_SMALLINT        对应PHOENIX里的UNSIGNED_SMALLINT      2字节
  7. $UNSIGNED_FLOAT           对应PHOENIX里的UNSIGNED_FLOAT         4字节
  8. $UNSIGNED_DOUBLE          对应PHOENIX里的UNSIGNED_DOUBLE        8字节
  9. $INTEGER                  对应PHOENIX里的INTEGER                4字节
  10. $BIGINT                   对应PHOENIX里的BIGINT                 8字节
  11. $TINYINT                  对应PHOENIX里的TINYINT                1字节
  12. $SMALLINT                 对应PHOENIX里的SMALLINT               2字节
  13. $FLOAT                    对应PHOENIX里的FLOAT                  4字节
  14. $DOUBLE                   对应PHOENIX里的DOUBLE                 8字节
  15. $BOOLEAN                  对应PHOENIX里的BOOLEAN                1字节
  16. $TIME                     对应PHOENIX里的TIME                   8字节
  17. $DATE                     对应PHOENIX里的DATE                   8字节
  18. $TIMESTAMP                对应PHOENIX里的TIMESTAMP              12字节
  19. $UNSIGNED_TIME            对应PHOENIX里的UNSIGNED_TIME          8字节
  20. $UNSIGNED_DATE            对应PHOENIX里的UNSIGNED_DATE          8字节
  21. $UNSIGNED_TIMESTAMP       对应PHOENIX里的UNSIGNED_TIMESTAMP     12字节
  22. $VARCHAR                  对应PHOENIX里的VARCHAR                动态长度
  23. $VARBINARY                对应PHOENIX里的VARBINARY              动态长度
  24. $DECIMAL                  对应PHOENIX里的DECIMAL                动态长度

如果不配置将以java对象原生类型默认映射转换

3.3 启动HBase数据同步

创建HBase表

在HBase shell中运行:

  1. create 'MYTEST.PERSON', {NAME=>'CF'}

启动canal-adapter启动器

  1. java -jar canal-adapter-launcher.jar

验证

修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log