技术是什么就是拿来玩的边玩边学才能成长得更快。之前已经给大家讲解了 MySQL 同步 ES 的几种方案下面就教大家如何通过 Canal将 MySQL 同步到 ES文章内容绝对妥妥干货本文会先讲解需要用到的基础知识然后再是软件安装最后就是实战部分。不 BB上文章目录01 基础知识1.1 主从复制原理MySQL 的主从复制是依赖于 binlog也就是记录 MySQL 上的所有变化并以二进制形式保存在磁盘上二进制日志文件。主从复制就是将 binlog 中的数据从主库传输到从库上一般这个过程是异步的即主库上的操作不会等待 binlog 同步地完成。详细流程如下主库写 binlog主库的更新 SQL(update、insert、delete) 被写到 binlog主库发送 binlog主库创建一个 log dump 线程来发送 binlog 给从库从库写 relay log从库在连接到主节点时会创建一个 IO 线程以请求主库更新的 binlog并且把接收到的 binlog 信息写入一个叫做 relay log 的日志文件从库回放从库还会创建一个 SQL 线程读取 relay log 中的内容并且在从库中做回放最终实现主从的一致性。1.2 Cannel 基础Canel 是一款常用的数据同步工具其原理是基于 Binlog 订阅的方式实现模拟一个 MySQL Slave 订阅 Binlog 日志从而实现 CDCChange Data Capture将已提交的更改发送到下游。主要流程如下Canal 服务端向 MySQL 的 master 节点传输 dump 协议MySQL 的 master 节点接收到 dump 请求后推送 Binlog 日志给 Canal 服务端解析 Binlog 对象原始为 byte 流转成 Json 格式Canal 客户端通过 TCP 协议或 MQ 形式监听 Canal 服务端同步数据到 ES。下面是 Cannel 执行的核心流程其中 Binlog Parser 主要负责 Binlog 的提取、解析和推送EventSink 负责数据的过滤 、路由和加工仅作了解即可。02 软件下载安装我的电脑是 Macos-x64所以后面的软件安装都是基于这个。2.1 Java JDK官网https://www.oracle.com/java/technologies/downloads/JDK 版本11.0.19由于 Canal 和 ES 的安装都强依赖 JDK所以这里有必要先说明。前方高能这里有坑如果你选的版本不对ES 安装可能会失败然后 Canal 同步数据到 ES 时也会出现很多诡异的问题。2.2 MySQLMySQL 大家应该都安装了这里需要打开 MySQL 的 BinLog。我是 Mac主要新建一个 my.cnf 文件然后再重启 MySQL。这里重启 MySQL我搞了半天BinLog 开启后会看到 BinLog 日志。然后需要创建一个账号账号和密码都是 Cannal给后面 Canal 使用。GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canallocalhost IDENTIFIED BY canal ;2.3 Canal官网https://github.com/alibaba/canal/releases版本v1.1.6下载 canal.adapter 和 canal.deployer 两个就可以canal.deployer相当于 canal 的服务端启动它才可以在客户端接收数据库变更信息。canal.adapter增加客户端数据落地的适配及启动功能(当 deployer 接收到消息后会根据不同的目标源做适配比如是 es 目标源适配和 hbase 适配等等)。备注canal.admin 为 canal提供整体配置管理、节点运维等面向运维的功能提供相对友好的 WebUI 操作界面方便更多用户快速和安全的操作我这边使用的是单机的因此就没有下载安装大家也可以拉 source code 源码去研究下。2.4 ESES 官网https://www.elastic.co/cn/downloads/elasticsearchES 版本7.17.4Mac 安装 ES 非常简单brew install elasticsearch安装细节不赘述安装成功后输入以下网址http://localhost:9200/?pretty2.5 Kibana下载网址https://www.elastic.co/cn/downloads/past-releases#kibana版本7.14.0它是 ES 的界面化操作工具安装细节不赘述安装成功后输入以下网址http://localhost:5601/app/dev_tools#/console2.6 IK 分词器下载网址https://github.com/medcl/elasticsearch-analysis-ik/releases/tag/v7.17.2版本v7.17.2它是 ES 的分词器安装细节不赘述安装成功后可以验证一下分词效果。2.7 小节MySQL 开启 BinLog这个不难主要观察是否有 BinLog 日志。Canal 的安装是最复杂的涉及到很多配置修改后面会讲解。最后是 ES Kibana IK 分词器这个其实也不难主要关注 ES 绑定的 JDK 版本03 Canal 配置3.1 canal.deployer 配置修改 conf—example 文件夹的 instance.properties 监听的数据库配置。这里主要修改的监听 MySQL 的 URL、用户名和密码。这里默认的账号密码就是 canal前面已经教大家如何创建了。3.2 canal.deployer 启动在 canal.deployer 中的 bin 文件下去启动命令 startup.sh这样就代表已经启动了我们可以去看下启动日志。上面 start successful 代表已经启动成功并且已经监听我的 MySQL 数据库。3.3 canal.adapte 配置Step1: 先把 adapter 下面的 bootstrap.yml全部注释掉否则会提示你 XX 表不存在这里坑了我好惨。Step2: 再修改 adapter 的 application.yml 配置文件。这里的坑一般就是 mysql 的账号密码不对或者给的 es 链接没有http://前缀这些都是我过踩的坑。Step3: 修改我们在 application.yml 中配置的目标数据源 es7 文件夹内容。由于我们这里是对表 article 进行去监听因此我们在 es7 文件夹中去创建 article.yml 文件。由于我们需要把技术派项目中的文章查询功能改造成 ES 的查询方式所以我们就把技术派的文章表 article同步到 ES 中。yml文件配置如下dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 destination: example # canal的instance或者MQ的topic groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据 esMapping: _index: article # es 的索引名称 _id: _id # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配 sql: SELECT t.id AS _id,t.id,t.user_id,t.article_type,t.title,t.short_title,t.picture, t.summary,t.category_id,t.source,t.source_url,t.offical_stat,t.topping_stat, t.cream_stat,t.status,t.deleted,t.create_time,t.update_time FROM article t # sql映射 commitBatch: 1 # 提交批大小Step4: 在 Kibana 中创建 ES 的 article 索引代码如下PUT /article { mappings : { properties : { id : { type : integer }, user_id : { type : integer }, article_type : { type : integer }, title : { type : text, analyzer: ik_max_word }, short_title : { type : text, analyzer: ik_max_word }, picture : { type : text, analyzer: ik_max_word }, summary : { type : text, analyzer: ik_max_word }, category_id : { type : integer }, source : { type : integer }, source_url : { type : text, analyzer: ik_max_word }, offical_stat : { type : integer }, topping_stat : { type : integer }, cream_stat : { type : integer }, status : { type : integer }, deleted : { type : integer }, create_time : { type : date }, update_time : { type : date } } } }3.4 canal.adapte 启动我们看下启动日志上面没有任何报错并且已经启动了 8081 端口说明已经启动成功此时我们就可以操作了。04 数据同步实战4.1 全量同步在开始 adapter 之后我们应该先来一把全量数据同步在源码中提供了一个接口进行全量同步命令如下curl http://127.0.0.1:8081/etl/es7/article.yml -X POST上面就是执行同步成功后提示已经导入 10 条。4.2 增量同步增量数据就是当我在 MySQL 中 update、delete 和 insert 时那么 ES 中数据也会对应发生变化我下面演示下修改日志打印如下ES查询结果如下上面结果中说明 ES 已经更改成功。05 总结我们再回顾一下整体执行流程写到这里就结束了是不是满满的干货呢基本是手把手教你如何将 MySQL 同步到 ES不仅是增量同步还包括全量同步如果你的项目也需要用到该场景基本可以直接照搬。