chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Updated ClientAdapter (markdown)

agapple
2018-11-15 22:39:57 +08:00
parent b604ae0b61
commit b9be11f58f
1 changed files with 60 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
ClientAdapter.md

@@ -14,9 +14,8 @@ canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能,
* MySQL版本 ：5.7.18
* HBase版本: Apache HBase 1.1.2, 若和服务端版本不一致可自行替换客户端HBase依赖
## 一、适配器启动器
client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载
## 一、适配器整体结构
client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载, 目前所有支持的适配器都放置在plugin目录下
启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为:
```
@@ -39,9 +38,9 @@ client-adapter分为适配器和启动器两部分, 适配器为多个fat jar,
```
以上目录结构最终会打包成 canal-adapter-*.tar.gz 压缩包
## 二、启动器
### 2.1 启动器配置 application.yml
#### canal相关配置部分说明
## 二、适配器配置介绍
### 2.1 总配置文件 application.yml
#### adapter定义配置部分
```
canal.conf:
canalServerHost: 127.0.0.1:11111 # 对应单机模式下的canal server的ip:port
@@ -50,7 +49,7 @@ canal.conf:
flatMessage: true # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
canalInstances: # canal实例组, 如果是tcp模式可配置此项
- instance: example # 对应canal destination
groups: # 对应适配器分组, 分组间的适配器并行运行
groups: # 对应适配器分组, 分组间的适配器并行运行
- outAdapters: # 适配器列表, 分组内的适配串行运行
- name: logger # 适配器SPI名
- name: hbase
@@ -66,7 +65,12 @@ canal.conf:
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、数据同步回查等使用
@@ -77,35 +81,12 @@ adapter.conf:
adapterConfigs: # 适配器内部配置列表
- hbase/mytest_person2.yml # 类型/配置文件名, 这里示例就是对应HBase适配器hbase目录下的mytest_person2.yml文件
```
## 2.2 同步管理REST接口
#### 2.2.1 查询所有订阅同步的canal destination或MQ topic
```
curl http://127.0.0.1:8081/destinations
```
#### 2.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中的数据同步开关, 如果是单机模式则使用本地锁来控制开关
#### 2.2.3 数据同步开关状态
```
curl http://127.0.0.1:8081/syncSwitch/example
```
查看指定 canal destination/MQ topic 的数据同步开关状态
#### 2.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"
```
导入数据到指定类型的库
#### 2.2.5 查看相关库总数据
```
curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
```
### 2.3 启动canal-adapter示例
#### 2.3.1 启动canal server (单机模式), 参考: [Canal QuickStart](https://github.com/alibaba/canal/wiki/QuickStart)
#### 2.3.2 修改conf/application.yml为:
## 三、适配器启动
### 3.1 启动canal-adapter示例
#### 3.1.1 启动canal server (单机模式), 参考: [Canal QuickStart](https://github.com/alibaba/canal/wiki/QuickStart)
#### 3.1.2 修改conf/application.yml为:
```
server:
port: 8081
@@ -127,201 +108,66 @@ canal.conf:
- outAdapters:
- name: logger
```
启动
#### 3.1.3 启动
```
bin/startup.sh
```
## 三、HBase适配器
### 3.1 修改启动器配置: application.yml
### 3.2 adapter管理REST接口
#### 3.2.1 查询所有订阅同步的canal destination或MQ topic
```
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
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代表开启
canal.conf:
canalServerHost: 127.0.0.1:11111
flatMessage: true
srcDataSources:
defaultDS:
url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
username: root
password: 121212
canalInstances:
- instance: example
adapterGroups:
- outAdapters:
- name: hbase
properties:
hbase.zookeeper.quorum: slave1
hbase.zookeeper.property.clientPort: 2181
zookeeper.znode.parent: /hbase
注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关
#### 3.2.3 数据同步开关状态
```
adapter将会自动加载 conf/hbase 下的所有.yml结尾的配置文件
### 3.2 适配器表映射文件
修改 conf/hbase/mytest_person.yml文件:
curl http://127.0.0.1:8081/syncSwitch/example
```
dataSourceKey: defaultDS # 对应application.yml中的datasourceConfigs下的配置
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:
查看指定 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即可
...
columns:
id: ROWKE$STRING
...
type: TYPE$BYTE
...
- outAdapters:
- name: logger
```
类型转换涉及到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.3 启动HBase数据同步
#### 创建HBase表
在HBase shell中运行:
```
create 'MYTEST.PERSON', {NAME=>'CF'}
```
#### 启动canal-adapter启动器
```
bin/startup.sh
```
#### 验证
修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log
### Hbase适配器
## 四、关系型数据库适配器
同步HBase配置 : [[Sync-HBase]]
RDB adapter 用于适配mysql到任意关系型数据库(需支持jdbc)的数据同步及导入
### RDB适配器
### 4.1 修改启动器配置: application.yml, 这里以oracle目标库为例
```
server:
port: 8081
logging:
level:
com.alibaba.otter.canal.client.adapter.rdb: DEBUG
......
同步关系型数据库配置 : [[Sync-RDB]]
支持的数据库列表:
1. MySQL
2. Oracle
3. Postgress
4. SQLServer
canal.conf:
canalServerHost: 127.0.0.1:11111
srcDataSources:
defaultDS:
url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
username: root
password: 121212
canalInstances:
- instance: example
groups:
- outAdapters:
- name: rdb # 指定为rdb类型同步
key: oracle1 # 指定adapter的唯一key, 与表映射配置中outerAdapterKey对应
properties:
jdbc.driverClassName: oracle.jdbc.OracleDriver # jdbc驱动名, jdbc的jar包需要自行放致lib目录下
jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc url
jdbc.username: mytest # jdbc username
jdbc.password: m121212 # jdbc password
threads: 5 # 并行执行的线程数, 默认为1
commitSize: 3000 # 批次提交的最大行数
```
其中 outAdapter 的配置: name统一为rdb, key为对应的数据源的唯一标识需和下面的表映射文件中的outerAdapterKey对应, properties为目标库jdb的相关参数
adapter将会自动加载 conf/rdb 下的所有.yml结尾的表映射配置文件
使用了JDBC driver,理论上支持绝大部分的关系型数据库
### 4.2 适配器表映射文件
修改 conf/rdb/mytest_user.yml文件:
```
dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值
destination: example # cannal的instance或者MQ的topic
outerAdapterKey: oracle1 # adapter key, 对应上面配置outAdapters中的key
concurrent: true # 是否按主键hase并行同步, 并行同步的表必须保证主键不会更改及主键不能为其他同步表的外键!!
dbMapping:
database: mytest # 源数据源的database/shcema
table: user # 源数据源表名
targetTable: mytest.tb_user # 目标数据源的库名.表名
targetPk: # 主键映射
id: id # 如果是复合主键可以换行映射多个
# mapAll: true # 是否整表映射, 要求源表和目标表字段名一模一样 (如果targetColumns也配置了映射,则以targetColumns配置为准)
targetColumns: # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填
id:
name:
role_id:
c_time:
test1:
```
导入的类型以目标表的元类型为准, 将自动转换
### ES适配器
### 4.3 启动RDB数据同步
#### 将目标库的jdbc jar包放入lib文件夹, 这里放入ojdbc6.jar
### MongoDB适配器
#### 启动canal-adapter启动器
```
bin/startup.sh
```
#### 验证
修改mysql mytest.user表的数据, 将会自动同步到Oracle的MYTEST.TB_USER表下面, 并会打出DML的log
### Redis适配器