cannl同步mysql数据到es中

canal组件介绍

canal-admin
(非必须但推荐使用)：为canal提供整体配置管理、节点运维等面向运维的功能，提供相对友好的WebUI操作界面，方便更多用户快速和安全的操作。

canal-server：
服务端，从mysql读取binlog日志获取增量日志，可以通过tcp、kafka、RocketMQ等方式与客户端通信；通过zookeeper搭建集群。

canal-adapter
客户端，根据canal-server获取的增量日志执行适配到其他诸如elasticsearch、redis、mysql等端，实现数据同步。

本次实验环境

首先我们需要下载canal的各个组件canal-server、canal-adapter、canal-admin，下载地址：https://github.com/alibaba/canal/releases

es及相关工具的部署可参考Elasticsearch详解及部署

1、开启mysql的binlog

使用canal-server需要先准备mysql，对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式

vi /etc/my.cnf

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

配置完成后重启mysql，并查询是否配置生效：ON就是开启

systemctl restart mysqld

可以进入数据库再次查看下

mysql> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | ON |
+---------------+-------+
1 row in set (0.01 sec)

mysql> show variables like 'binlog_format%';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+
1 row in set (0.00 sec)

2、mysql用户数据准备

 创建一个canal用户并对其进行授权

CREATE USER canal IDENTIFIED BY 'Cjz123456.';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

创建一个测试用户数据库及表

create database canal;
USE canal;
CREATE TABLE product  (
  id bigint(20) NOT NULL AUTO_INCREMENT,
  title varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  sub_title varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  price decimal(10, 2) NULL DEFAULT NULL,
  pic varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (id) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

3、创建存放canal的目录　　

mkdir /usr/local/canal-server

mkdir /usr/local/canal-adpter　　

4、部署canal-server

tar -zxvf canal.deployer-1.1.6-SNAPSHOT.tar.gz -C /usr/local/canal-server

vim /usr/local/canal-server/conf/example/instance.properties

# 需要同步数据的MySQL地址
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# 用于同步数据的数据库账号
canal.instance.dbUsername=canal
# 用于同步数据的数据库密码
canal.instance.dbPassword=Cjz123456.
# 数据库连接编码
canal.instance.connectionCharset = UTF-8
# 需要订阅binlog的表过滤正则表达式
canal.instance.filter.regex=.*\\..*

cd /usr/local/canal-server/bin

./startup.sh

查看日志查看是否正常启动

cd /usr/local/canal-server/logs/

tail -f example/example.log

tail -f canal/canal.log

5、部署canal-adapter

tar -zxvf canal.adapter-1.1.6-SNAPSHOT.tar.gz -C /usr/local/canal-adpter/

cd /usr/local/canal-adpter/conf

vim 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 # 客户端的模式，可选tcp kafka rocketMQ
  flatMessage: true # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
  zookeeperHosts:    # 对应集群模式下的zk地址
  syncBatchSize: 1000 # 每次同步的批数量
  retries: 0 # 重试次数, -1为无限重试
  timeout: # 同步超时时间, 单位毫秒
  accessKey:
  secretKey:
  consumerProperties:
    canal.tcp.server.host: 127.0.0.1:11111 #设置canal-server的地址
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://192.168.111.129:3306/canal?useUnicode=true&charaterEncoding=utf-8&useSSL=false
      username: canal
      password: Cjz123456.
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7
        hosts: http://192.168.111.129:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: rest # or rest
          security.auth: elastic:elastic #  only used for rest mode
          cluster.name: my-es

修改 canal-adapter/conf/es7/mytest_user.yml 文件，用于配置MySQL中的表与Elasticsearch中索引的映射关系

vim es7/mytest_user.yml

dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值
destination: example  # canal的instance或者MQ的topic
groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:
  _index: canal_product # es 的索引名称
  _id: _id  # 将Mysql表里的id对应上es上的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
  sql: "SELECT
         p.id AS _id,
         p.title,
         p.sub_title,
         p.price,
         p.pic
        FROM
         product p"        # sql映射
  etlCondition: "where p.id>={}"   #etl的条件参数
  commitBatch: 3000   # 提交批大小

cd /usr/local/canal-adpter/bin/

./startup.sh

查看日志看是否正常启动 

cd /usr/local/canal-adpter/logs/adapter/

tail -f adapter.log

6、创建ES索引

可以直接通过Kibana页面进行创建，不难看出，索引的属性对应着就是我们表的属性(Mysql表id与es的_id做了对应，所以这里不需要设置)

PUT canal_product
{
  "mappings": {
    "properties": {
      "title": {
        "type": "text"
      },
      "sub_title": {
        "type": "text"
      },
      "pic": {
        "type": "text"
      },
      "price": {
        "type": "double"
      }
    }
  }
}

7、验证

canal初始化是不会全量同步数据的，所以我们需要登录mysql，插入一条数据

INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 15, '小米8', ' 全面屏游戏智能手机 6GB+64GB', 1999.00, NULL );

创建完成后在Kibana上查看即可发现数据已经同步

GET canal_product/_search

在elasticsearch-head上查看也是，可以看出，我们的表id变成了es的_id

在Mysql上修改数据看看

UPDATE product SET title='小米10' WHERE id=15;

 可以看到我们修改的数据也进行了同步

该文章参考于：https://blog.csdn.net/zh1998wx/article/details/123101442