使用canal同步MySQL数据到Elasticsearch(ES)
目录
- 1、功能及使用场景
- 1.1、功能介绍
- 1.2、使用场景
- 2、需求引入
- 3、canal文件下载及准备
- 3.1 下载文件
- 3.2 准备文件
- 4、deployer安装及效果测试
- 4.1、deployer 配置修改
- 4.1.1 准备
- 4.1.2 修改连接数据库信息
- 4.2 启动deployer
- 4.3 测试deployer效果
- 4.3.1 在本地电脑新建普通maven工程
- 4.3.2 新建ClientSimple类
- 5、adapter安装及效果测试
- 5.1 修改配置
- 5.1.1 修改启动器配置: application.yml
- 5.1.2 修改 conf/es/mytest_user.yml文件
- 5.2 启动adapter
- 5.3 效果测试
- 6、全量数据导入
1、功能及使用场景
1.1、功能介绍
canal是阿里巴巴开源的mysql数据传输组件,基于mysql binlog,提供了准确、实时的数据传输服务。有关binlog介绍,参见binlog介绍。
以下来自官方GitHub介绍。GitHub地址
canal [kə’næl],译意为水道/管道/沟渠,主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费
早期阿里巴巴因为杭州和美国双机房部署,存在跨机房同步的业务需求,实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始,业务逐步尝试数据库日志解析获取增量变更进行同步,由此衍生出了大量的数据库增量订阅和消费业务。
基于日志增量订阅和消费的业务包括
- 数据库镜像
- 数据库实时备份
- 索引构建和实时维护(拆分异构索引、倒排索引等)
- 业务 cache 刷新
- 带业务逻辑的增量数据处理
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x
1.2、使用场景
根据官方文档,目标系统支持MySQL、kafka、elasticsearch、hbase、rocketMQ、pulsar等。
本文主要介绍MySQL同步到Elasticsearch的使用。
2、需求引入
整个过程用一个示例来介绍canal的安装使用。
存在如下两张表(一对多关系),用户信息表和用户权限表,需要将用户信息以及权限同步到es
user_info(用户信息表):
id | name | role_id |
---|---|---|
1 | 张三 | 1 |
2 | 李四 | 2 |
3 | 王大锤 | 2 |
role(权限表):
role_id | role_name |
---|---|
1 | 管理员 |
2 | 测试员 |
查询sql:
SELECT a.id AS _id, a.name, a.role_id, b.role_name
FROM
user_info a
LEFT JOIN role b
ON
b.role_id = a.role_id
查询结果:
_id | name | role_id | role_name |
---|---|---|---|
3 | 王大锤 | 2 | 测试员 |
1 | 张三 | 1 | 管理员 |
2 | 李四 | 1 | 管理员 |
对应的es索引结构:
{"user_index": {"mappings": {"_doc": {"properties": {"name": {"type": "text"},"role_id": {"type": "long"},"role_name": {"type": "text"}}}}}
}
3、canal文件下载及准备
3.1 下载文件
官方GitHub下载地址:https://github.com/alibaba/canal/releases
下载canal服务端(canal.deployer-1.1.4.tar.gz)和客户端(canal.adapter-1.1.4.tar.gz),如下图。
下载完成如下
3.2 准备文件
命令运行完成后,进入adapter和deployer可以看到如下结构(忽略我本机的.DS_Store)
adapter:
deployer:
4、deployer安装及效果测试
4.1、deployer 配置修改
4.1.1 准备
(针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步)
对于自建 MySQL , 需要先开启 Binlog 写入功能,配置 binlog-format 为 ROW 模式,my.cnf 中配置如下
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant,下面新建了canal账号
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;
4.1.2 修改连接数据库信息
vi conf/example/instance.properties
修改如下标红信息
4.2 启动deployer
在deployer目录运行启动脚本
sh bin/startup.sh
- 查看 server 日志
tail -f logs/canal/canal.log
- 查看 instance 的日志
tail -f logs/example/example.log
看到如上日志,标志启动成功
4.3 测试deployer效果
(4.3 步骤可跳过,主要为验证deployer端效果)
4.3.1 在本地电脑新建普通maven工程
pom文件,依赖如下pom
<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.0</version>
</dependency>
4.3.2 新建ClientSimple类
粘贴测试代码(下面链接页面上的ClientSimple代码)
https://github.com/alibaba/canal/wiki/ClientExample
将圈红ip改为部署adapter的ip,然后直接启动此main方法
启动完成看到如下日志:
日志会循环打印count
此时触发数据库变更
user_info表变更前
user_info表变更后,新加了一条名叫逻辑的记录
可在日志处观察到变更信息,标志着deployer监听mysql binlog变更成功
5、adapter安装及效果测试
canal adapter 的 Elastic Search 版本支持6.x.x以上
官方文档地址
https://github.com/alibaba/canal/wiki/Sync-ES
5.1 修改配置
5.1.1 修改启动器配置: application.yml
进adapter/conf目录
vi application.yml
修改如下配置,注意缩进格式,yml文件严格缩进格式,格式错误会引起启动失败问题。
mode使用rest模式,测试使用transport会出问题,目前没找到原因,下图mode还没更改
5.1.2 修改 conf/es/mytest_user.yml文件
adapter将会自动加载 conf/es 下的所有.yml结尾的配置文件
不需要的文件可删除,只配置需要的yml文件
修改如下配置,esMapping信息,包括 index,type,sql,其中sql尽量保证不要换行,在文本编辑器中编辑成一行,在粘贴进去,否则可能会出问题(还是由于yml格式问题)
此处的sql的字段别名即是es字段名,不写别名,默认原名即是es字段名,有关详细说明,可参见官方文档https://github.com/alibaba/canal/wiki/Sync-ES
etlCondition 可以注释掉,我们默认任何条件都同步
5.2 启动adapter
启动命令 sh bin/startup.sh
查看日志:
tail -f logs/adapter/adapter.log
观察如下日志,即表示启动成功
5.3 效果测试
canal是一个MySQL增量订阅组件,所以不支持数据的初始化
我们需要在数据库触发变更,才能将数据同步到es
变更前:
es数据->在kibana查询对应数据,可以看到右侧数据为空
mysql数据→使用查询sql,查询到数据如下
下面,我们进行数据变更:
比如将张三名字变成张六,MySQL数据如下:
此时,es数据在MySQL数据变更同时,es数据相应变更,如下,可以观察到,数据变更已经成功从MySQL同步到elasticsearch
需要注意的是,由于我们只变更了user_info表,所以此处只同步了user_info表的name和id字段,role_name字段,并没有同步,只有role_name字段变更时,才会被同步,所以实际使用时,要先做好数据初始化工作。
同时,可在logs/adapter/adapter.log观察到数据变更日志
至此,全部操作完成
6、全量数据导入
由于canal只支持增量导入,所以官方adapter提供了全量导入手动触发的功能,
canal全表同步(etl功能,手动触发)
参见源码:
/*** ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST** @param type 类型 hbase, es* @param task 任务名对应配置文件名 mytest_person2.yml* @param params etl where条件参数, 为空全部导入*/
@PostMapping("/etl/{type}/{task}")
public EtlResult etl(@PathVariable String type, @PathVariable String task,@RequestParam(name = "params", required = false) String params) {return etl(type, null, task, params);
}
只需要发送http命令即可
例:如下,ip和端口是部署canal adapter的IP和端口,然后类型选es,后面在跟上对应的yml配置文件即可
curl http://127.0.0.1:8081/etl/es/mytest_person2.yml -X POST
导入成功提示
使用canal同步MySQL数据到Elasticsearch(ES)相关推荐
- 使用canal实时同步MySQL数据到Elasticsearch
使用canal实时同步MySQL数据到Elasticsearch 搭建环境 安装 elasticsearch 安装 kibana 下载和安装canal 1.下载canal 2.配置MySQL 3.配置 ...
- Elasticsearch7.9集群部署,head插件,canal同步mysql数据到es,亲自测试,无坑
Elasticsearch集群部署 1.服务器规划 10.4.7.11 node1 10.4.7.12 node2 10.4.7.13 node3 1. 集群相关 一个运行中的 Elastics ...
- canal deployer+canal adapter自动同步MySQL数据到ElasticSearch
一.安装Elastic Search 6.8 1.下载 Elastic Search 6.8资源下载 2.部署环境配置 创建用户(es不能使用root启动,否则报错) adduser elastic ...
- Logstash 安装及简单实用(同步MySql数据到Elasticsearch)
Logstash是一款轻量级的日志搜集处理框架,可以方便的把分散的.多样化的日志搜集起来,并进行自定义的处理,然后传输到指定的位置,比如某个服务器或者文件 Windows环境: 1.下载logstas ...
- Flink 最佳实践之使用 Canal 同步 MySQL 数据至 TiDB
摘要:本文将介绍如何将 MySQL 中的数据,通过 Binlog + Canal 的形式导入到 Kafka 中,继而被 Flink 消费的案例.内容包括: 背景介绍 环境介绍 部署 TiDB Clus ...
- canal同步mysql数据到rocketmq集群
rockermq多主多从异步复制部署参考 canal github 早期阿里巴巴因为杭州和美国双机房部署,存在跨机房同步的业务需求,实现方式主要是基于业务 trigger 获取增量变更.从 2010 ...
- canal同步mysql数据
canal简介 canal官网文档 canal [kə'næl],译意为水道/管道/沟渠,主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费 早期阿里巴巴因为杭州和美国双机房部署, ...
- canal同步mysql到kafka_使用Canal同步MySQL数据到Kafka 得到的数据中sql字段无值-问答-阿里云开发者社区-阿里云...
这个应该跟你的binlog记录模式有关系,binlog有3中模式,ROW(行模式), Statement(语句模式), Mixed(混合模式)三种模式的用法如下: ROW(行模式):记录那条数据修改了 ...
- logstash同步mysql数据到elasticsearch,动态模板
后台运行logstash Last login: Thu Jul 23 17:50:26 2020 [root@localhost ~]# nohup /home/essoft/logstash-6 ...
最新文章
- 原来这才是 Kafka!(多图+深入)
- 关于程序中数据库报错“父项未找到”
- 网站防火墙探测工具Wafw00f
- JSR303(Bean Validation 1.0)
- 使用SSH+SFTP操作终端全解析,告别XShell
- [概统]本科二年级 概率论与数理统计 第七讲 期望、方差与相关性
- TImm 笔记: 训练模型
- 产品经理能力与职责表:你达标了几项?
- Visual Studio 2019 + MFC——配置MFC
- mysql视图子查询_mysql创建视图不能包涵子查询的解决办法。View's SELECT contains a subquery in the FROM clause...
- php负责传递数据,php传递数据的方法有哪些
- 防止U盘中病毒的小技巧
- 再学 GDI+[20]: TGPTextureBrush 与 TWrapMode
- mfc 子窗体 按钮不触发_PIE二次开发在子窗体中选择主窗体中的文件
- 2003系统如何搭建ftp服务器配置,WINDOWSSERVER2003系统架设FTP服务器配置方法.pdf
- 简单的使用Python处理Excel数据文件
- ubuntu安装flash_如何在Ubuntu Linux上安装最新版本的Flash
- Mex-hdu4747(DP)
- windows不能更改密码
- 国科大学习资料--人工智能原理与算法-第七次作业解析(学长整理)
热门文章
- 【小白学习C++ 教程】二十一、C++ 中的STL容器Arrays和vector
- 八、Linux 常用 Shell 命令,控制台的快捷键以及 Shell 编程(中)
- 三十五、深入Java中的泛型(下篇)
- ICLR 2022 | 香侬科技提出基于图神经网络的语义理解模型,获单项满分
- 视频预测领域有哪些最新研究进展?不妨看看这几篇顶会论文
- Keras梯度累积优化器:用时间换取效果
- AAAI 2018论文解读 | 基于文档级问答任务的新注意力模型
- HDU 1069 Monkey and Banana 最长上升子序列进阶(动态规划)
- poj2002 STL set
- 数据挖掘实战(三):特征工程-二手车交易价格预测