使用Canal同步数据到ES

一、Canal概述

1、Canal是什么？

Canal是阿里巴巴开源的一个组件，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。canal的介绍，在github 上的官方文档介绍的很好，我这边就不介绍了。感兴趣的查看git地址：https://github.com/alibaba/canal

2、Canal版本

Canal 1.1.4版本，迎来最重要的WebUI能力，引入canal-admin工程，支持面向WebUI的canal动态管理能力，支持配置、任务、日志等在线白屏运维能力，具体文档:Canal Admin Guide。

二、Canal同步到ES

1、服务器规划

本地测试准备2台服务器

服务器	部署的服务
a-lf-bigdata	mysql、canal-server
b-lf-bigdata	mysql、canal-admin、canal-adapter、es、kibana

2、安装mysql

A. 分别在两台机器安装mysql

（1）安装MySQL的yum仓库

yum -y localinstall https://dev.mysql.com/get/mysql80-community-release-el7-3.noarch.rpm

（2）安装MySQL

yum -y install mysql-community-server

（3）设置为开机启动

systemctl enable mysqld

（4）启动MySQL

systemctl start mysqld

（5）查看MySQL状态

systemctl status mysqld

（6）查看root临时密码

grep 'temporary password' /var/log/mysqld.log

（7）修改root密码

mysql -uroot -p
ALTER USER 'root'@'localhost' IDENTIFIED BY 'Root_12root';
SHOW VARIABLES LIKE 'validate_password%';
set global validate_password.policy=0;
set global validate_password.length=1;
ALTER USER 'root'@'localhost' IDENTIFIED BY 'root%123';
exit

B. MySQL创建用户

1、在a-lf-bigdata的MySQL创建采集数据的用户

mysql -uroot -p
set global validate_password.policy=0;
set global validate_password.length=1;
CREATE USER canal IDENTIFIED BY '2wsxVFR_';
-- GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;
exit

在a-lf-bigdata的MySQL开启Binlog格式

vi /etc/my.cnf

增加如下配置

server-id=1
log-bin=mysql-bin
binlog-format=ROW
binlog-ignore-db=information_schema
binlog-ignore-db=mysql
binlog-ignore-db=performance_schema
binlog-ignore-db=sys

log-bin用于指定binlog日志文件名前缀，默认存储在/var/lib/mysql 目录下。
server-id用于标识唯一的数据库，不能和别的服务器重复，建议使用ip的最后一段，默认值也不可以。
binlog-ignore-db:表示同步的时候忽略的数据库。
binlog-do-db:指定需要同步的数据库(如果没有此项，表示同步所有的库)

登录mysql查看：

mysql -uroot -p
show master status;

如图，binlog就成功开启了。

禁用explicit_defaults_for_timestamp

mysql -uroot -p
SHOW VARIABLES LIKE '%explicit_defaults_for_timestamp%';
set persist explicit_defaults_for_timestamp=0;
SHOW VARIABLES LIKE '%explicit_defaults_for_timestamp%';

重启MySQL

systemctl status mysqld

2、在b-lf-bigdata的MySQL创建用户

mysql -uroot -p
set global validate_password.policy=0;
set global validate_password.length=1;
CREATE USER canaladmin IDENTIFIED BY '2wsxVFR_';
GRANT ALL ON canal_manager.* TO 'canaladmin'@'%';
FLUSH PRIVILEGES;
exit

3、安装ES

1、下载安装包

下载ES：

curl -L -O https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.4.0-linux-x86_64.tar.gz
curl -L -O https://artifacts.elastic.co/downloads/kibana/kibana-7.4.0-linux-x86_64.tar.gz

2、安装ES

（1）为服务器创建一个普通用户

useradd hadoop
passwd hadoop

然后赋予sudo权限

（2）安装ES

解压ES

cd /data/liufei/es
tar -zxf elasticsearch-7.4.0-linux-x86_64.tar.gz
ln -s elasticsearch-7.4.0 elasticsearch

配置环境变量

# 编辑配置文件
vi /etc/profile
# 新增ES配置
#elasticsearch
export ES_HOME=/data/liufei/es/elasticsearch
export PATH=${ES_HOME}/bin:$PATH
# 是配置生效
source /etc/profile

基础配置

调整最大虚拟内存

# 编辑配置文件
vi /etc/sysctl.conf
# 增加配置
vm.max_map_count=262144

保存退出后执行命令使配置生效

sysctl -p

修改权限

将ES目录的所有者赋予给hadoop用户

chown -R hadoop:hadoop elasticsearch
chown -R hadoop:hadoop elasticsearch-7.4.0

切换到hadoop用户

su hadoop

配置ES

# 编辑配置文件
vi $ES_HOME/config/elasticsearch.yml# 修改配置
network.host: 0.0.0.0
discovery.seed_hosts: ["b-lf-bigdata"]

启动ES

$ES_HOME/bin/elasticsearch -d

确认ES启动成功

curl http://b-lf-bigdata:9200

（2）安装kibana

解压kibana

cd /data/liufei/es
tar xzvf kibana-7.4.0-linux-x86_64.tar.gz
ln -s kibana-7.4.0-linux-x86_64 kibana

配置环境变量

# 编辑配置文件
vi /etc/profile
# 新增ES配置
#kibana
export KIBANA_HOME=/data/liufei/es/kibana
export PATH=${KIBANA_HOME}/bin:$PATH
# 是配置生效
source /etc/profile

配置Kibana

# 编辑配置文件
vi $KIBANA_HOME/config/kibana.yml# 修改配置
server.host: "0.0.0.0"
elasticsearch.hosts: ["http://b-lf-bigdata:9200"]

启动Kibana

nohup $KIBANA_HOME/bin/kibana > $KIBANA_HOME/kibana.out 2>&1 &

查看Kibana状态

http://b-lf-bigdata:5601

新建索引

put /test_test
{"mappings": {"properties": {"name": {"type": "text"}, "age": {"type": "integer"},"modified_time": {"type": "date"} }}
}

4、安装Canal

（1）下载Canal

Canal下载地址：https://github.com/alibaba/canal/releases
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gz
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gz

（2）安装Canal-admin

解压

tar -zxf canal.admin-1.1.5.tar.gz
cd canal-admin

目录结构如下

配置环境变量

# 编辑配置文件
vi /etc/profile# 增加配置
#canal-admin
export CANAL_ADMIN_HOME=/data/liufei/cacal/canal-admin
export PATH=${CANAL_ADMIN_HOME}/bin:$PATH# 使环境变量生效
source /etc/profile

修改配置

vi $CANAL_ADMIN_HOME/conf/application.yml# 修改配置如下
server:port: 8089
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8
spring.datasource:address: b-lf-bigdata:3306database: canal_managerusername: canaladminpassword: 2wsxVFR_driver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?
useUnicode=true&characterEncoding=UTF-
8&useSSL=false&allowPublicKeyRetrieval=truehikari:maximum-pool-size: 30minimum-idle: 1
#这里指的是canal-server和canal-admin双线通信的用户名密码 canal:adminUser: adminadminPasswd: 123456

替换MySQL驱动包

因为我们使用的是MySQL 5.7，所以需要使用8的驱动

初始化数据库

mysql -uroot -p
source /data/liufei/canal/canal-admin/conf/canal_manager.sql

启动canal-admin

sh $CANAL_ADMIN_HOME/bin/startup.sh

查看日志：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lomrs253-1631081115831)(/Users/juzi/Library/Application Support/typora-user-images/image-20210908132642239.png)]

Canal-admin启动成功，访问地址

http://b-lf-bigdata:8089

（3）安装Canal-server

解压

tar -zxvf canal.deployer-1.1.5.tar.gz
cd canal-server

配置环境变量

vi /etc/profile# 新增配置
#canal-server
export CANAL_SERVER_HOME=/home/hadoop/app/canal-server
export PATH=${CANAL_SERVER_HOME}/bin:$PATH# 使环境变量生效
source /etc/profile

修改配置

vi canal.properties# register ip
canal.register.ip = b-lf-bigdata# canal admin config
canal.admin.manager = b-lf-bigdata:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name =

更换MySQL 驱动包
启动

$CANAL_SERVER_HOME/bin/startup.sh

查看日志

（4）配置Instance

登录canal_admin
进入Instance 管理页，新建instance

载入模版

修改模版

canal.instance.master.address=a-lf-bigdata:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=2wsxVFR_

启动instance，查看日志

（5）创建调试数据库

在a-lf-bigdata上操作

mysql -uroot -p# 创建数据
CREATE DATABASE IF NOT EXISTS test DEFAULT CHARSET utf8 COLLATE utf8_general_ci;
use test;
DROP TABLE IF EXISTS `test`;
CREATE TABLE `test` (`uid` INT UNSIGNED AUTO_INCREMENT,`name` VARCHAR(100) NOT NULL,`age` int(3) DEFAULT NULL,`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP,PRIMARY KEY (`uid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

（6）安装ClientAdapter

解压

tar -zxf canal.adapter-1.1.5.tar.gz
cd canal-adapter

配置环境变量

vi /etc/profile# 新增配置
#canal-adapter
export CANAL_ADAPTER_HOME=/data/liufei/cacal/canal-adapter
export PATH=${CANAL_ADAPTER_HOME}/bin:$PATH# 使环境变量生效
source /etc/profile

修改服务配置

vi $CANAL_ADAPTER_HOME/conf/application.ymlserver:port: 8081
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8default-property-inclusion: non_nullcanal.conf:mode: tcp #tcp kafka rocketMQ rabbitMQflatMessage: truezookeeperHosts:syncBatchSize: 1000retries: 0timeout:accessKey:secretKey:consumerProperties:# canal tcp consumercanal.tcp.server.host: a-lf-bigdata:11111canal.tcp.zookeeper.hosts:canal.tcp.batch.size: 500canal.tcp.username:canal.tcp.password:srcDataSources:defaultDS:url: jdbc:mysql://a-lf-bigdata:3306/test?useUnicode=trueusername: canalpassword: 2wsxVFR_canalAdapters:- instance: test_to_es # canal instance Name or mq topic namegroups:- groupId: g1outerAdapters:- 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: es6hosts: b-lf-bigdata:9300 # 127.0.0.1:9200 for rest modeproperties:mode: transport # or rest
#          # security.auth: test:123456 #  only used for rest modecluster.name: es
#        - name: kudu
#          key: kudu
#          properties:
#            kudu.master.address: 127.0.0.1 # ',' split multi address

修改es6 目录下的配置，新建test_to_es.yml

dataSourceKey: defaultDS
destination: test_to_es
groupId: g1
esMapping:_index: test_test_type: _doc_id: _idupsert: truesql: "select a.uid as _id, a.name, a.age, a.modified_time from test a"commitBatch: 2

替换MySQL驱动包
启动adapter

$CANAL_ADAPTER_HOME/bin/startup.sh
$CANAL_ADAPTER_HOME/bin/stop.sh
$CANAL_ADAPTER_HOME/bin/restart.sh

（7）测试

新增数据

INSERT INTO test.test (name, age) VALUES ("张三",20);
INSERT INTO test.test (name, age) VALUES ("李四",21);
INSERT INTO test.test (name, age) VALUES ("王五",22);
INSERT INTO test.test (name, age) VALUES ("赵六",23);
INSERT INTO test.test (name, age) VALUES ("马七",24);

查看es