二、适配器配置介绍
2.1 总配置文件 application.yml
adapter定义配置部分
canal.conf: canalServerHost: 127.0.0.1:11111 # 对应单机模式下的canal server的ip:port zookeeperHosts: slave1:2181 # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准 mqServers: slave1:6667 #or rocketmq # kafka或rocketMQ地址, 与canalServerHost不能并存 flatMessage: true # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效 batchSize: 50 # 每次获取数据的批大小, 单位为K syncBatchSize: 1000 # 每次同步的批数量 retries: 0 # 重试次数, -1为无限重试 timeout: # 同步超时时间, 单位毫秒 mode: tcp # kafka rocketMQ # canal client的模式: tcp kafka rocketMQ srcDataSources: # 源数据库 defaultDS: # 自定义名称 url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true # jdbc url username: root # jdbc 账号 password: 121212 # jdbc 密码 canalAdapters: # 适配器列表 - instance: example # canal 实例名或者 MQ topic 名 groups: # 分组列表 - groupId: g1 # 分组id, 如果是MQ模式将用到该值 outerAdapters: # 分组内适配器列表 - name: logger # 日志打印适配器 ......
说明:
- 一份数据可以被多个group同时消费, 多个group之间会是一个并行执行, 一个group内部是一个串行执行多个outerAdapters, 比如例子中logger和hbase
- 目前client adapter数据订阅的方式支持两种，直连canal server 或者 订阅kafka/RocketMQ的消息
2.2 使用远程配置(Mysql)
可以使用远程配置中心(Mysql，可扩展)作为统一配置管理
2.1.1 创建mysql schema
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配置
canal: manager: jdbc: url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8 username: root password: 121212
可以将本地application.yml文件和其他子配置文件删除或清空， 启动工程将自动从远程加载配置
修改mysql中的配置信息后会自动刷新到本地动态加载相应的实例或者应用
三、适配器启动
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 batchSize: 500 syncBatchSize: 1000 retries: 0 timeout: mode: tcp canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - name: logger
3.1.3 启动
bin/startup.sh
3.2 adapter管理REST接口
3.2.1 查询所有订阅同步的canal instance或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 instance/MQ topic 进行开关操作. off代表关闭, instance/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启
注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关
3.2.3 数据同步开关状态
curl http://127.0.0.1:8081/syncSwitch/example
查看指定 canal instance/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"
导入数据到指定类型的库, 如果params参数为空则全表导入, 参数对应的查询条件在配置中的etlCondition指定
3.2.5 查看相关库总数据
curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
适配器列表
logger适配器
最简单的处理, 将受到的变更事件通过日志打印的方式进行输出, 如配置所示, 只需要定义name: logger即可 ... outerAdapters: - name: logger
Hbase适配器
Sync HBase
rewerma edited this page on 1 Feb 2019 · 5 revisions
背景
canal 1.1.1版本之后, 内置增加客户端数据同步功能, Client适配器整体介绍: ClientAdapter
HBase适配器
1. 修改启动器配置: application.yml
canal.conf: canalServerHost: 127.0.0.1:11111 batchSize: 500 syncBatchSize: 1000 retries: 0 timeout: mode: tcp # kafka rocketMQ srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true username: root password: 121212 canalAdapters: - instance: example # canal instance Name or mq topic name groups: - groupId: g1 outerAdapters: - name: hbase properties: hbase.zookeeper.quorum: 127.0.0.1 hbase.zookeeper.property.clientPort: 2181 zookeeper.znode.parent: /hbase
注意：adapter将会自动加载 conf/hbase 下的所有.yml结尾的配置文件
2. HBase表映射文件
修改 conf/hbase/mytest_person.yml文件:
dataSourceKey: defaultDS # 对应application.yml中的datasourceConfigs下的配置 destination: example # 对应tcp模式下的canal instance或者MQ模式下的topic groupId: # 对应MQ模式下的groupId, 只会同步对应groupId的数据 hbaseMapping: # mysql--HBase的单表映射配置 mode: STRING # HBase中的存储类型, 默认统一存为String, 可选: #PHOENIX #NATIVE #STRING # NATIVE: 以java类型为主, PHOENIX: 将类型转换为Phoenix对应的类型 destination: example # 对应 canal destination/MQ topic 名称 database: mytest # 数据库名/schema名 table: person # 表名 hbaseTable: MYTEST.PERSON # HBase表名 family: CF # 默认统一Column Family名称 uppercaseQualifier: true # 字段名转大写, 默认为true commitBatch: 3000 # 批量提交的大小, ETL中用到 #rowKey: id,type # 复合字段rowKey不能和columns中的rowKey并存 # 复合rowKey会以 '|' 分隔 columns: # 字段映射, 如果不配置将自动映射所有字段, # 并取第一个字段为rowKey, HBase字段名以mysql字段名为主 id: ROWKE name: CF:NAME email: EMAIL # 如果column family为默认CF, 则可以省略 type: # 如果HBase字段和mysql字段名一致, 则可以省略 c_time: birthday:
注意: 如果涉及到类型转换,可以如下形式:
... columns: id: ROWKE$STRING ... type: TYPE$BYTE ...
类型转换涉及到Java类型和Phoenix类型两种, 分别定义如下:
#Java 类型转换, 对应配置 mode: NATIVE $DEFAULT $STRING $INTEGER $LONG $SHORT $BOOLEAN $FLOAT $DOUBLE $BIGDECIMAL $DATE $BYTE $BYTES #Phoenix 类型转换, 对应配置 mode: PHOENIX $DEFAULT 对应PHOENIX里的VARCHAR $UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节 $UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节 $UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节 $UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节 $UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节 $UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节 $INTEGER 对应PHOENIX里的INTEGER 4字节 $BIGINT 对应PHOENIX里的BIGINT 8字节 $TINYINT 对应PHOENIX里的TINYINT 1字节 $SMALLINT 对应PHOENIX里的SMALLINT 2字节 $FLOAT 对应PHOENIX里的FLOAT 4字节 $DOUBLE 对应PHOENIX里的DOUBLE 8字节 $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节 $TIME 对应PHOENIX里的TIME 8字节 $DATE 对应PHOENIX里的DATE 8字节 $TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节 $UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节 $UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节 $UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节 $VARCHAR 对应PHOENIX里的VARCHAR 动态长度 $VARBINARY 对应PHOENIX里的VARBINARY 动态长度 $DECIMAL 对应PHOENIX里的DECIMAL 动态长度
如果不配置将以java对象原生类型默认映射转换
3. 启动HBase数据同步
3.1 创建HBase表
在HBase shell中运行:
create 'MYTEST.PERSON', {NAME=>'CF'}
3.2 启动canal-adapter启动器
bin/startup.sh
3.4 验证
修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log
RDB适配器
同步关系型数据库配置 : Sync-RDB
目前内置支持的数据库列表:
- MySQL
- Oracle
- PostgresSQL
- SQLServer
使用了JDBC driver,理论上支持绝大部分的关系型数据库
ES适配器
Sync ES
rewerma edited this page on 26 Sep 2019 · 16 revisions
背景
canal 1.1.1版本之后, 内置增加客户端数据同步功能, Client适配器整体介绍: ClientAdapter
canal adapter 的 Elastic Search 版本支持6.x.x以上, 如需其它版本的es可替换依赖重新编译client-adapter.elasticsearch模块
ElasticSearch适配器
1 修改启动器配置: application.yml
canal.conf: canalServerHost: 127.0.0.1:11111 batchSize: 500 syncBatchSize: 1000 retries: 0 timeout: mode: tcp srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true username: root password: 121212 canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - key: exampleKey name: es6 # or es7 hosts: 127.0.0.1:9300 # es 集群地址, 逗号分隔 properties: mode: transport # or rest # 可指定transport模式或者rest模式 # security.auth: test:123456 # only used for rest mode cluster.name: elasticsearch # es cluster name
adapter将会自动加载 conf/es 下的所有.yml结尾的配置文件
2 适配器表映射文件
修改 conf/es/mytest_user.yml文件:
dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 outerAdapterKey: exampleKey # 对应application.yml中es配置的key destination: example # cannal的instance或者MQ的topic groupId: # 对应MQ模式下的groupId, 只会同步对应groupId的数据 esMapping: _index: mytest_user # es 的索引名称 _type: _doc # es 的type名称, es7下无需配置此项 _id: _id # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配 # pk: id # 如果不需要_id, 则需要指定一个属性为主键属性 # sql映射 sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name, a.c_time as _c_time, c.labels as _labels from user a left join role b on b.id=a.role_id left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id" # objFields: # _labels: array:; # 数组或者对象属性, array:; 代表以;字段里面是以;分隔的 # _obj: object # json对象 etlCondition: "where a.c_time>='{0}'" # etl 的条件参数 commitBatch: 3000 # 提交批大小
sql映射说明:
sql支持多表关联自由组合, 但是有一定的限制:
- 主表不能为子查询语句
- 只能使用left outer join即最左表一定要是主表
- 关联从表如果是子查询不能有多张表
- 主sql中不能有where查询条件(从表子查询中可以有where条件但是不推荐, 可能会造成数据同步的不一致, 比如修改了where条件中的字段内容)
- 关联条件只允许主外键的’='操作不能出现其他常量判断比如: on a.role_id=b.id and b.statues=1
- 关联条件必须要有一个字段出现在主查询语句中比如: on a.role_id=b.id 其中的 a.role_id 或者 b.id 必须出现在主select语句中
Elastic Search的mapping 属性与sql的查询值将一一对应(不支持 select *), 比如: select a.id as _id, a.name, a.email as _email from user, 其中name将映射到es mapping的name field, _email将 映射到mapping的_email field, 这里以别名(如果有别名)作为最终的映射字段. 这里的_id可以填写到配置文件的 _id: _id映射.
2.1.单表映射索引示例sql:
select a.id as _id, a.name, a.role_id, a.c_time from user a
该sql对应的es mapping示例:
{ "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "c_time": { "type": "date" } } } } } }
2.2.单表映射索引示例sql带函数或运算操作:
select a.id as _id, concat(a.name,'_test') as name, a.role_id+10000 as role_id, a.c_time from user a
函数字段后必须跟上别名, 该sql对应的es mapping示例:
{ "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "c_time": { "type": "date" } } } } } }
2.3.多表映射(一对一, 多对一)索引示例sql:
select a.id as _id, a.name, a.role_id, b.role_name, a.c_time from user a left join role b on b.id = a.role_id
注:这里join操作只能是left outer join, 第一张表必须为主表!!
该sql对应的es mapping示例:
{ "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "role_name": { "type": "text" }, "c_time": { "type": "date" } } } } } }
2.4.多表映射(一对多)索引示例sql:
select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id
注:left join 后的子查询只允许一张表, 即子查询中不能再包含子查询或者关联!!
该sql对应的es mapping示例:
{ "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "c_time": { "type": "date" }, "labels": { "type": "text" } } } } } }
2.5.其它类型的sql示例:
- geo type
select ... concat(IFNULL(a.latitude, 0), ',', IFNULL(a.longitude, 0)) AS location, ...
- 复合主键
select concat(a.id,'_',b.type) as _id, ... from user a left join role b on b.id=a.role_id
- 数组字段
select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id
配置中使用:
objFields: labels: array:;
- 对象字段
select a.id as _id, a.name, a.role_id, c.labels, a.c_time, a.description from user a
配置中使用:
objFields: description: object
其中a.description字段内容为json字符串
- 父子文档索引
es/customer.yml
...... esMapping: _index: customer _type: _doc _id: id relations: customer_order: name: customer sql: "select t.id, t.name, t.email from customer t"
es/order.yml
esMapping: _index: customer _type: _doc _id: _id relations: customer_order: name: order parent: customer_id sql: "select concat('oid_', t.id) as _id, t.customer_id, t.id as order_id, t.serial_code as order_serial, t.c_time as order_time from biz_order t" skips: - customer_id
mapping示例:
{ "mappings":{ "_doc":{ "properties":{ "id": { "type": "long" }, "name": { "type": "text" }, "email": { "type": "text" }, "order_id": { "type": "long" }, "order_serial": { "type": "text" }, "order_time": { "type": "date" }, "customer_order":{ "type":"join", "relations":{ "customer":"order" } } } } } }
3 启动ES数据同步
启动canal-adapter启动器
bin/startup.sh
