elasticSearch|canal1.15 增量同步 mysql5.7 数据到ElasticSearch7.14.0 mysql|大数据|canal|增量同步|el

前言： canal由java开发，需要JDK1.8+ 环境,没有则安装 yum install java

通过开源源码可以知道目前canal目前最高支持为7.3.0，截止2021-11-04 最新为7.15.X。
本地搭建的ElasticSearch版本为7.14.0。测试兼容没问题

文章图片

1:MySQL开启binlog 修改mysql配置文件 /etc/my.cnf
[mysqld]下增加：

log-bin=/usr/local/mysql/binlogs/mysql-bin #开启 binlogbinlog-format=ROW #选择 ROW 模式,必须为ROW行模式server_id=1 #配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

2：MySQL验证binlog -- 查看binlog是否正确启动 row

show variables like 'binlog_format%';

-- 查看log_bin 是否为ON

show variables like 'log_bin%';

elasticSearch|canal1.15 增量同步 mysql5.7 数据到ElasticSearch7.14.0

文章图片

3：部署Canal-deployer服务端 3.1 下载canal.deployer1.1.5

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.deployer-1.1.5-SNAPSHOT.tar.gz

3.2 解压

tar -zxvf canal.deployer-1.1.5-SNAPSHOT.tar.gz

3.3 跳转配置目录

cd canal.deployer-1.1.5/conf/example

3.4 修改数据库连接信息

vim instance.properties

文章图片

3.5 启动系统

canal.deployer-1.1.5/bin/startup.sh

3.6 查看进程是否启动

ss -anplt | grep java

deployer默认监听三个端口,11110、11111、11112，
adapter启动会依赖deployer的11111进程端口。

文章图片

如未启动，查看日志进行排查

tail -f 300 canal.deployer-1.1.5/logs/canal/canal.log

文章图片

4：部署Canal-adapter客户端 4.1 下载canal.adapter-1.1.5

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.adapter-1.1.5-SNAPSHOT.tar.gz

4.2 解压

tar -zxvf canal.adapter-1.1.5-SNAPSHOT.tar.gz

4.3 修改application.yum配置信息

vim adapter/conf/application.yml

文章图片

USER: HOSTNAME%%.*: PWD/#$HOME/\~: server: port: 9091 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: mode: tcp flatMessage: true zookeeperHosts: syncBatchSize: 1000 retries: 0 timeout: accessKey: secretKey: consumerProperties: # canal tcp consumer canal.tcp.server.host: 127.0.0.1:11111 canal.tcp.zookeeper.hosts: canal.tcp.batch.size: 500 canal.tcp.username: canal.tcp.password: srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf8&useAffectedRows=false&allowMultiQueries=true&serverTimezone=GMT%2B8&useSSL=false&allowMultiQueries=true&useUnicode=true username: root password: root canalAdapters: - instance: example # canal instance Name or mq topic name groups: - groupId: g1 outerAdapters: - name: logger - name: es7 hosts: http://127.0.0.1:9200 properties: mode: rest security.auth: elastic:pusamm1314.. cluster.name: elasticsearch

4.4 启动canal.adapter-1.1.5

adapter1.1.5/bin/startup.sh

4.5 查看日志是否启动（遇到各种坑，我启动了4小时才成功。先启动这个，可以忽略elasticsearch 索引配置问题。启动成功再配置索引信息）

tail -f 300 adapter1.1.5/logs/adapter/adapter.log

Canal部署过程中的错误可以简单参考下Canal部署过程中的错误_Asker.J的博客-CSDN博客
4.6 es7配置需要先创建elasticSearch 索引和数据库表信息
4.6.1elasticSearch 创建索引 canal

文章图片

{ "settings": { "number_of_shards": 5, "analysis": { "tokenizer": { "ngram_tokenizer": { "type": "nGram", "min_gram": 1, "max_gram": 1, "token_chars": [ "letter", "digit" ] } }, "analyzer": { "license_plate_analyzer": { "tokenizer": "ngram_tokenizer", "filter": [ "uppercase" ] } } } }, "mappings": { "properties": { "id": { "type": "long" }, "name": { "type": "text", "analyzer": "license_plate_analyzer", "search_analyzer": "license_plate_analyzer" }, "sex": { "type": "text", "analyzer": "license_plate_analyzer", "search_analyzer": "license_plate_analyzer" }, "age": { "type": "text", "analyzer": "license_plate_analyzer", "search_analyzer": "license_plate_analyzer" } } } }

4.6.2 创建数据库表canal

create table `canal` ( id int(11) not null auto_increment, name varchar(20) null comment '名称', sexvarchar(2) null comment '性别', ageint null comment '年龄', primary key (`ID`) ) engine=innodb auto_increment=3 default charset=utf8;

4.6.3 创建索引对应配置文件
cd adapter1.1.5/conf/es7（原目录下默认三个demo。可参考，不能与elasticsearch中所有的索引名称一样）
touch canal.yml（创建，可直接用vim代替）
vim canal.yml

文章图片

dataSourceKey: defaultDS destination: example groupId: esMapping: _index: canal _type: _doc _id: _id upsert: true sql: "select a.id as _id,a.sex,a.name from canal a" commitBatch: 3000

4.6.4 重启查看是否正常启动

adapter1.1.5/bin/stop.shadapter1.1.5/bin/startup.shtail -f 300 adapter1.1.5/logs/adapter/adapter.log

5：插入/修改/删除数据测试

insert into canal(id,name,sex,age) values(null,'欧阳修','男',12); insert into canal(id,name,sex,age) values(null,'李白','男',15); insert into canal(id,name,sex,age) values(null,'韩愈','男',18); insert into canal(id,name,sex,age) values(null,'柳宗元','男',45); insert into canal(id,name,sex,age) values(null,'苏轼','男',26); insert into canal(id,name,sex,age) values(null,'苏辙','男',22); insert into canal(id,name,sex,age) values(null,'曾巩','男',98); update canal set name='欧阳修11',sex='女' where age=12; update canal set name='李白',sex='女' where name='李白'; delete from canal where name = '李白'

/adapter.log