- 使用Canal实现es数据实时同步
image.png
canal ：github地址：https://github.com/alibaba/canal/
原理很简单：数据库开启bin-log日志，calan伪装成mysql的slave 读取bin-log日志
image.png
官方的说明：
canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)
最新Canal版本为 1.1.4 支持 ES7
image.png
(1)、 下载canal项目（git命令也可以 - - 随意.）
image.png
(2)、解压
image.png
(3)、打开终端 cd到项目目录
image.png
mvn clean install
说明：没有配置 maven 的 自己百度配置一下 - - .
安装完成后：打开target文件夹
image.png
就可以看到 tar包.
把
canal.deployer-1.1.5-SNAPSHOT.tar.gz
canal.adapter-1.1.5-SNAPSHOT.tar.gz
canal.admin-1.1.5-SNAPSHOT.tar.gz
上传到 centos .并解压.
image.png
mkdir canal-xxxx
tar -zxf canal.xxxx-1.1.5-SNAPSHOT.tar.gz -C /usr/soft/canal-xxxx
先新建目录，然后再解压到相应目录. xxxx 是不同的项目.
cd canal-deployer/conf/example
修改instance.properties
vim instance.properties
image.png
还有一些其他配置：稍后再说
image.png
最后切换到 bin目录 start.sh 即可
image.png
查看日志启动成功
日志在logs下面：
tail -200f example/example.log
canal1.1.4提供 web-ui
官方说明：
canal 1.1.4版本，迎来最重要的WebUI能力，引入canal-admin工程，
支持面向WebUI的canal动态管理能力，支持配置、任务、日志等在线白屏运维能力，
具体文档：[Canal Admin Guide](https://github.com/alibaba/canal/wiki/Canal-Admin-Guide)
image.png
主要配置在 conf中
首先：创建数据库canal_manager
image.png
然后编辑application.yml 修改成自己的数据库信息
image.png
修改instance-template.properties
image.png
最后切换到 bin目录 启动即可
访问路径：
http://192.168.0.196:8089
image.png
账号：admin
密码：123456
image.png
详情参考：https://github.com/alibaba/canal/wiki/Canal-Admin-Guide
进入 canal-adapter/conf 目录
image.png
这里需要配置 application.yml 和es7（因为我们使用的es是7.0.1版本）
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: 192.168.0.196:11111 #canal默认地址 可以在 canal-deployer 项目中配置
# zookeeperHosts: slave1:2181
# mqServers: 127.0.0.1:9092 #or rocketmq
# flatMessage: true
batchSize: 500
syncBatchSize: 1000
retries: 0
timeout:
accessKey:
secretKey:
username:
password:
vhost:
srcDataSources:
defaultDS:
url: jdbc:mysql://192.168.0.201:3306/khfooddb?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: mysql1
# properties:
# jdbc.driverClassName: com.mysql.jdbc.Driver
# jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
# jdbc.username: root
# jdbc.password: 121212
# - 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: es7 #es版本选择 es6 or es7
hosts: 192.168.0.196: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
编辑 es7/mytest_user.yml
为什么是他呢。。。
因为：
image.png
image.png
相对应，也可以自定义。。。
vim es7/mytest_user.yml
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: coremerchant_index # es 的index
# _type: _doc
_id: _id
upsert: true
# pk: id
sql: "select a.id as _id,a.id,a.name,a.status,concat(IFNULL(a.latitude, 0), ',', IFNULL(a.longitude, 0)) AS location from core_merchant a"
# objFields:
# _labels: array:;
# etlCondition: "where t.c_time>={}"
commitBatch: 3000
最后切换到 bin 目录 startup.sh
然后 tailf ../logs/adapter/adapter.log
查看启动日志
最后进行测试- -
自己数据库对应的表 INSERT INTO 一条数据，看看查看效果.
