canal实现MySQL和ES同步实践

前言

canal是阿里开源的数据同步工具，基于binlog可以将数据库同步到其他各类数据库中，目标数据库支持mysql,postgresql,oracle,redis,MQ,ES等。

一、 基本原理

1、主从复制原理

MySQL的主从复制是依赖于binlog，也就是记录MySQL上所有操作的逻辑日志保存在磁盘上。主从复制就是将binlog中的数据从主库传输到从库上，一般这个过程是异步的。

原理示意图：

详细流程：

• 主库写binlog：主库的更新SQL(create、update、insert、delete等)被写入到binlog；
• 主库发送binlog：主库创建一个log dump线程来发送binlog给从库；
• 从库写relay log：(从库在连接到主节点时会创建一个IO线程，以请求主库更新的binlog)，并且把binlog信息写入一个叫做relay log的日志文件；
• 从库回放：从库还会创建一个SQL线程读取relay log中的内容，并且在从库中回访，最终实现主从一致性。

2、Canal介绍

主要流程：

• Canal 服务端向 MySQL 的 master 节点传输dump协议；
• MySQL 的 master 节点接收到 dump 请求后推送 binlog 日志给 Canal 服务端，解析 binlog 对象（原始为byte流）转成 Json 格式；
• Canal 客户端通过 TCP 协议或 MQ 形式监听 Canal 服务端，同步数据到ES。
  

二、实操

这篇文章我将以“将 MySQL 中 coding 数据库的 article 表同步到 ES ”为需求举例
看一下article表的结构（这里展示主要是为了展示后文的 yml文件 和 ES索引 是怎么回事）：

1、软件下载

个人电脑：Windows 11
JDK版本：jdk 11.0.21

1.1 MySQL

（默认MySQL都已经安装了，登录 mysql 按如下执行）

• 在 mysql 命令行输入show variables like 'log_bin;'查看是否开启binlog，如下图显式 NO 为开启
  
• 创建一个 canal 用户，专门用于 canal 同步数据库使用

# 注意大小写
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

1.2 ES

• 安装地址：https://www.elastic.co/cn/downloads/past-releases#elasticsearch
• 版本：7.17.2
  
  1、点击下载
  2、将下载后的压缩包解压到一文件夹（路径最好不要含有中文或空格）
  3、双击bin/elasticsearch.bat启动
  
  4、检查是否成功，输入http://localhost:9200/?pretty
  

1.3 Kibana

1、安装地址：https://www.elastic.co/cn/downloads/past-releases#kibana
2、版本：7.14.0
3、解压安装到指定文件夹
4、双击 D:\ProgramFiles\kibana-7.14.0-windows-x86_64\bin\kibana.bat 启动
5、输入 http://localhost:5601 检验是否启动成功

6、在 Kibana 中创建ES索引（可以通过PostMan创建，看下图示意）

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"
            }
        }
    }
}

PostMan执行成功后返回200。

1.4 ik

1、下载地址：https://gitcode.com/medcl/elasticsearch-analysis-ik/tags/v7.17.2?utm_source=csdn_github_accelerator&isLogin=1
2、版本：7.17.2 （ik的版本必须与ES版本一致）
3、在 ES 目录下的 plugins 中创建一个文件夹ik（ES的安装目录最好不要有空格就是因为我安装的时候如果有空格 ik 启动的时候会报错）
4、将 ik 压缩包解压到 plugins/ik 目录下，解压后的目录结构如下

5、重启elasticsearch、kibana
6、访问 http://localhost:5601/app/dev_tools#/console，输入方框中的内容，点击运行，出现右侧的结果，就是说明安装成功了。

1.5 Canal

下载配置Canal的时候真的遇到了很多乱七八糟奇奇怪怪的问题，我的建议是面向百度编程搜索无解的时候不如静下来分析一下日志。

1、下载地址：https://github.com/alibaba/canal/releases
2、版本：1.1.7

先把框中这三个压缩包下载下来解压到文件夹：比如

canal.deployer：相当于canal的服务端，启动它才可以在客户端接收数据库变更信息
canal.adapter：增加客户端数据落地的适配以及启动功能（当 deployer 收到消息后，会根据不同的目录源做适配，比如 es 目标源适配等。看需求，目标源是Redis就不用这个。）

2、canal 配置

2.1 canal.deployer 配置

修改 canal.deployer-1.1.7\conf\example\instance.properties 文件

2.2 canal.deployer 启动

1、双击 “D:\ProgramFiles\canal.deployer-1.1.7\bin\startup.bat” 启动
2、在 “D:\ProgramFiles\canal.deployer-1.1.7\logs\canal\canal.log” 中查看启动日志，存在类似于下图的字样说明启动成功，也可能会在日志中看到binlog转成Json后的sql逻辑

3、如果之前同步过，想要重新做全量同步，那么需要删除 “D:\ProgramFiles\canal.deployer-1.1.7\conf\example\meta.dat” 文件，这个文件记录上次同步的时间和 binlog 位置。

2.3 source源码配置

我上边的 canal-canal-1.1.7 文件夹就是 source 解压之后
1、用 IDEA 打开 canal-canal-1.1.7 项目，等待依赖加载完毕，编译成功
2、在 “canal-canal-1.1.7\client-adapter\es7x\pom.xml” 中增加依赖

        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>client-adapter.common</artifactId>
            <version>${project.version}</version>
            <scope>provided</scope>
        </dependency>

3、在 “canal-canal-1.1.7\pom.xml” 中增加如下片段，主要是打包跳过测试，避免报错

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>${maven-surefire.version}</version>
                <configuration>
                    <useSystemClassLoader>true</useSystemClassLoader>
                    <forkMode>once</forkMode>
                    <argLine>${argline} ${jacocoArgLine}</argLine>
                    <systemProperties>
                        <!-- common shared -->
                    </systemProperties>
                    <skipTests>true</skipTests>    <!--默认关掉单元测试 -->
                </configuration>
            </plugin>

4、在maven插件中重新打包

5、打包成功后将打包后的产物 "D:\ProgramFiles\canal-canal-1.1.7\client-adapter\es7x\target\client-adapter.es7x-1.1.7-jar-with-dependencies.jar" 替换 "D:\ProgramFiles\canal.adapter-1.1.7\plugin\client-adapter.es7x-1.1.7-jar-with-dependencies.jar" （注意是把 source 中的产物替换到 adapter 文件夹下）

2.4 canal.adapter配置

1、先将 “D:\ProgramFiles\canal.adapter-1.1.7\conf\bootstrap.yml” 文件中的全部内容注释（这个文件应该是给 canal.admin 用的，不注释掉的话启动时会提供XX表不存在）
2、修改 “D:\ProgramFiles\canal.adapter-1.1.7\conf\application.yml” 文件，重点看一下方框里的内容就是，其他默认的不用改

3、在 “D:\ProgramFiles\canal.adapter-1.1.7\conf\es7” 文件夹创建 article.yml 文件

文件内容如下：

datasourceKey: defaultDS #源数据源的key，对应上面配置的srcDatasources中的值
destination: example #canal的instance或者MO的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" # sq1映射
commitBatch: 1 # 提交批大小

“D:\ProgramFiles\canal.adapter-1.1.7\conf\es7” 解压后这个文件夹下应该会有3个默认的 yml 文件，最好删掉，不然启动的时候会报错

2.5 canal.adapter 启动

1、双击 “D:\ProgramFiles\canal.adapter-1.1.7\bin\startup.bat” 启动
2、到 “D:\ProgramFiles\canal.adapter-1.1.7\logs\adapter\adapter.log” 中查看 日志，有这样的显示说明启动成功，并且执行同步操作了

3、日志的最后能看到同步的条数和时间

4、也可以通过 curl 进行全量同步操作

curl http://127.0.0.1:8081/etl/es7/article.yml -X POST

5、在 Kibana 中查看，成功。

三、Q&A

1、好多教程里没有 source 的部分，启动adapter的就会出现问题，先把source 的部分配置好再试下；

2、source源码在IDEA中编译一直失败

可能是这个地方没有改成你自己的仓库，也可能就是网络不好。

3、启动adapter时报错：Failed to bind properties under ‘es-mapping’ to com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig$ESMapping

用记事本打开 startup.bat 文件，classPath 分号前后的内容顺序换一下，下面是改好后的样子

4、忘记截图了，看日志大概意思是说“找不到XXX索引”

找不到的索引是 canal.adapter 文件夹解压的时候 canal.adapter-1.1.7\conf\es7 里自带的，把自带的yml文件删除就好了文章来源地址https://www.toymoban.com/news/detail-851141.html

5、如果想起来其他问题会再补充，启动 adapter 时一些问题其实不太容易直接搜出来的，这个时候自己多看一下 adapter 的日志，尝试理解问题现象思考原因。

到了这里，关于canal实现MySQL和ES同步实践的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！