項目的需求是mysql到mysql
一、首先需要安裝canal-server，之前的博客也簡單寫過，這就跳過了。
二、配置canal.adapter
- 下載canal.adapter 解壓 https://github.com/alibaba/canal/releases
- 編輯
application.yml
server:
port: 8081
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
default-property-inclusion: non_null
canal.conf:
mode: tcp # kafka rocketMQ
canalServerHost: 127.0.0.1:11111
# zookeeperHosts: slave1:2181
# mqServers: 127.0.0.1:9092 #or rocketmq
# flatMessage: true
batchSize: 500
syncBatchSize: 1000
retries: 0
timeout:
accessKey:
secretKey:
srcDataSources:
demoSrc:
url: jdbc:mysql://192.168.1.104:3308/canal?useUnicode=true
username: root
password: root
canalAdapters:
- instance: example # canal instance Name or mq topic name
groups:
- groupId: g1
outerAdapters:
- name: logger
- name: rdb
key: demoDes
properties:
jdbc.driverClassName: com.mysql.jdbc.Driver
jdbc.url: jdbc:mysql://192.168.1.104:3308/test?useUnicode=true
jdbc.username: root
jdbc.password: root
# - name: rdb
# key: oracle1
# properties:
# jdbc.driverClassName: oracle.jdbc.OracleDriver
# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
# jdbc.username: mytest
# jdbc.password: m121212
# - name: rdb
# key: postgres1
# properties:
# jdbc.driverClassName: org.postgresql.Driver
# jdbc.url: jdbc:postgresql://localhost:5432/postgres
# jdbc.username: postgres
# jdbc.password: 121212
# threads: 1
# commitSize: 3000
# - name: hbase
# properties:
# hbase.zookeeper.quorum: 127.0.0.1
# hbase.zookeeper.property.clientPort: 2181
# zookeeper.znode.parent: /hbase
# - name: es
# hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode
# properties:
# mode: transport # or rest
# # security.auth: test:123456 # only used for rest mode
# cluster.name: elasticsearch
adapter.conf:
datasourceConfigs: # 數據源配置列表, 數據源將在適配器中用於ETL、數據同步回查等使用
defaultDS: # 數據源 dataSourceKey, 適配器中通過該值獲取指定數據源
url: jdbc:mysql://192.168.1.104:3308/canal?useUnicode=true
username: root
password: root
adapterConfigs: # 適配器內部配置列表
- rdb/book.yml
- rdb/cluster.yml
注意，最后這段是我參考官網的，寫多個文件對應多個表
dataSourceKey: demoSrc
destination: example
groupId: g1
outerAdapterKey: demoDes
concurrent: true
dbMapping:
database: canal
table: tb_book2
targetTable: test.tb_book2
targetPk:
id: id
mapAll: true
etlCondition: "where c_time>={}"
commitBatch: 3000 # 批量提交的大小
注意如果是多個主鍵可以換行對應
三、測試
其他的配置可以查看官網
基本說明
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管理接口, 運行目錄結構為:
- bin
restart.sh
startup.bat
startup.sh
stop.sh
- lib
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-launcher.tar.gz 壓縮包
二、啟動器
2.1 啟動器配置 application.yml
canal相關配置部分說明
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
適配器相關配置部分說明
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文件
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
2.3.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
啟動
bin/startup.sh
三、HBase適配器
3.1 修改啟動器配置: 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
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
adapter將會自動加載 conf/hbase 下的所有.yml結尾的配置文件
3.2 適配器表映射文件
修改 conf/hbase/mytest_person.yml文件:
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:
如果涉及到類型轉換,可以如下形式:
...
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.3 啟動HBase數據同步
創建HBase表
在HBase shell中運行:
create 'MYTEST.PERSON', {NAME=>'CF'}
啟動canal-adapter啟動器
bin/startup.sh
驗證
修改mysql mytest.person表的數據, 將會自動同步到HBase的MYTEST.PERSON表下面, 並會打出DML的log
四、關系型數據庫適配器
RDB adapter 用於適配mysql到任意關系型數據庫(需支持jdbc)的數據同步及導入
4.1 修改啟動器配置: application.yml, 這里以oracle目標庫為例
server:
port: 8081
logging:
level:
com.alibaba.otter.canal.client.adapter.rdb: DEBUG
......
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結尾的表映射配置文件
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:
導入的類型以目標表的元類型為准, 將自動轉換
4.3 啟動RDB數據同步
將目標庫的jdbc jar包放入lib文件夾, 這里放入ojdbc6.jar
啟動canal-adapter啟動器
bin/startup.sh
驗證
修改mysql mytest.user表的數據, 將會自動同步到Oracle的MYTEST.TB_USER表下面, 並會打出DML的log
五、ElasticSearch適配器
5.1 修改啟動器配置: application.yml
server:
port: 8081
logging:
level:
com.alibaba.otter.canal.client.adapter.es: 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
srcDataSources:
defaultDS:
url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
username: root
password: 121212
canalInstances:
- instance: example
adapterGroups:
- outAdapters:
- name: es
hosts: 127.0.0.1:9300 # es 集群地址, 逗號分隔
properties:
cluster.name: elasticsearch # es cluster name
adapter將會自動加載 conf/es 下的所有.yml結尾的配置文件
5.2 適配器表映射文件
修改 conf/es/mytest_user.yml文件:
dataSourceKey: defaultDS # 源數據源的key, 對應上面配置的srcDataSources中的值
destination: example # cannal的instance或者MQ的topic
esMapping:
_index: mytest_user # es 的索引名稱
_type: _doc # es 的doc名稱
_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: obj:{"test":"123"}
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映射.
5.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"
}
}
}
}
}
}
5.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"
}
}
}
}
}
}
5.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"
}
}
}
}
}
}
5.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"
}
}
}
}
}
}
5.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:;
5.3 啟動ES數據同步
啟動canal-adapter啟動器
bin/startup.sh
驗證
- 新增mysql mytest.user表的數據, 將會自動同步到es的mytest_user索引下面, 並會打出DML的log
- 修改mysql mytest.role表的role_name, 將會自動同步es的mytest_user索引中的role_name數據
- 新增或者修改mysql mytest.label表的label, 將會自動同步es的mytest_user索引中的labels數據