Canal-1.1.6推送数据至Elasticsearch7.16.2（整合版）

一、环境

1.系统环境

screenfetch

2.软件版本

• Mysql-5.7.36
• Elasticsearch7.16.2
• Kibana-7.16.2
• Canal.Admin-1.1.7-SNAPSHOT
• Canal.Deployer-1.1.7-SNAPSHOT
• Canal.Adapter-1.1.6（tcp模块插件用的1.1.7-SNAPSHOT版） (说明：adapter1.1.7存在es-mapping映射问题，adapter1.1.6存在tcp通讯问题故此缝合了一下，顺带修复了日期格式转换错误)alibaba官方下载地址：alibaba.canal.comcsdn阿坤缝合版地址：canal.all-1.6.7.tar.gz

./bin/mysql -S ./mysql.sock -P${port}  -uroot  -e"select version();"

curl --user elastic:Csdn@123 192.168.200.154:9200

二、部署

1.配置Canal-Admin
（注意：我用的自己缝合好的包，大家也可以根据需求没时间研究的话可以直接区下载已整合好的包：canal.all-1.6.7.tar.gz）

tar -xf canal.all-1.6.7.tar.gz 

创建canal_manager库

CREATE SCHEMA `canal_manager` DEFAULT CHARACTER SET utf8mb4 ;

配置数据库将canal_manager.sql表结构导入mysql

use canal_manager;source /srv/program/canal/admin/conf/canal_manager.sql
CREATE USER canaladmin IDENTIFIED BY 'Csdn@123.';
GRANT ALL ON canal_manager.* TO 'canaladmin'@'%';
FLUSH PRIVILEGES;

修改配置文件

vim conf/application.yml
...
spring.datasource:
  address: 127.0.0.1:3307
  database: canal_manager
  username: canaladmin
  password: Csdn@123.

启动

 ./bin/startup.sh

访问：http://ip:8089
用户：admin
密码：123456

2.配置Canal-Deployer
数据库开启binlog和row模式

#未开启的数据库vim my.cnf

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式server_id=1# 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复#已经开启的数据库
 SET SESSION binlog_format ='ROW';
 show variables like "%binlog_format%";

创建instance实例所需的数据库账户

#授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant
CREATE USER canal IDENTIFIED BY 'Csdn@123.';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

修改配置文件

vim conf/canal_local.properties

canal.ip =192.168.200.152

canal.admin.manager =192.168.200.152:8089
...

启动

./bin/startup.sh local

回admin平台查看（deployer服务会自动注册进去）

canal日志正常

instance实例日志正常（缝合包自带一个es7同步实例）

（注意：由于是本人整合过的安装包所以这些配置都是自带的，正常情况需要自己去创建instance实例并且绑定主机的）
3.配置Canal-Adapter
注释conf下的远程数据库链接地址bootstrap.yml

vim  conf/bootstrap.yml

修改配置文件

vim conf/application.yml
...
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 192.168.200.152:11111
   #canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://192.168.200.152:3307/jsjb_prod?useUnicode=true&characterEncoding=utf8
      username: canal
      password: Csdn@123.
  canalAdapters:
  - instance: jsjb_prod_sync_es # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7
        hosts: http://192.168.200.153:9200,http://192.168.200.154:9200,http://192.168.200.155:9200
        properties:
          mode: rest # or rest
          security.auth: elastic:Csdn@123 #  only used for rest mode
          cluster.name: devops-es

在conf创建es7映射配置文件order_complete_query_t.yml

vim conf/es7/order_complete_query_t.yml
dataSourceKey: defaultDS #数据源
destination: jsjb_prod_sync_es #instance实例名字
groupId: g1
esMapping:
  _index: order_complete_query_t  #es7索引名字
  _id: _id
  #  pk: ...

根据自己业务需要写sql就行了

(注意：先不要启动，需要先去es7创建对应的索引)
4.创建es7对应索引
从kibana开发者工具创建（映射索引涉及线上数据所以就不公布了）

启动adapter

 ./bin/startup.sh 

通过日志可以看到已经成功了

实践是检验真理的唯一标准！

三、测试

1.查询所有订阅同步的canal instance或MQ topic

curl http://127.0.0.1:8081/destinations

2.手动全部同步

curl http://localhost:8081/etl/es7/order_complete_query_t.yml -X POST

3.数据同步开关状态

curl http://127.0.0.1:8081/syncSwitch/example

4.查看相关库总数据

curl http://127.0.0.1:8081/count/es7/order_complete_query_t.yml

到此结束！