从零搭建：Canal实时数据管道打通MySQL与Elasticsearch

news/2025/2/13 12:40:39/

CanalMysql_BinlogElasticsearchcenter_0">
Canal实时同步Mysql Binlog至 Elasticsearch


文章目录

  • CanalMysql_BinlogElasticsearchcenter_0" rel="nofollow">
    Canal实时同步Mysql **Binlog**至**Elasticsearch**
      • 一. 环境准备
        • 1.环境检查
          • 检查`Mysql`是否开启`BinLog`
          • 开启Mysql Binlog
          • Java环境检查
        • 2.新建测试库和表
        • 3.新建Es索引
      • 二.**部署 Canal Server**
        • **2.1 解压安装包**
        • **2.2 配置 Canal Server**
        • **2.3 启动 Canal Server**
      • **三. 部署 Canal Adapter（同步到 Elasticsearch）**
        • **3.1 配置 Adapter**
        • **3.2 配置数据映射**
        • **3.3 启动 Adapter**
      • **4. 验证同步**
        • **4.1 插入测试数据到 MySQL**
        • **4.2 查询 Elasticsearch**

一. 环境准备

  • 操作系统：Linux（Ubuntu 20.04）
  • Java 环境：JDK 8+（建议 OpenJDK 11）
  • MySQL：已启用 Binlog（ROW 模式），并创建 Canal 用户
  • Elasticsearch：已部署（版本 7.x 或 8.x）
  • Canal 二进制包：从 Canal Release 下载 canal.deployer-1.1.8.tar.gz 和 canal.adapter-1.1.8.tar.gz
1.环境检查
  • 检查Mysql是否开启BinLog
#root账号执行
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

输出如下证明已经打开：

image-20250211103029832

创建 Canal 用户并授权：

#创建用户
CREATE USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'Password@123';
# 给新创建账户赋予从库权限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';# 刷新权限
FLUSH PRIVILEGES;

如果没打开BinLog可以通过如下方法打开：

  • 开启Mysql Binlog

修改my.cnf文件，加入如下内容：

log_bin=mysql-bin
binlog_format=ROW
binlog_expire_logs_seconds=172800
expire_logs_days=2

log_bin:启用二进制日志，日志文件会以 mysql-bin 为前缀，并依次生成日志文件（例如：mysql-bin.000001，mysql-bin.000002 等）。

binlog_format:设置使用的二进制日志格式，在 MySQL 8.0 版本中，binlog_format 的默认值已经变为 ROW。所以，即使你在配置文件中没有明确设置 binlog_format，MySQL 会默认使用 ROW 作为二进制日志格式。在较早的 MySQL 版本中默认值是 STATEMENT。

binlog_expire_logs_seconds=172800 和 expire_logs_days=2：这些配置设置了二进制日志的过期时间（默认情况下，MySQL 会保留二进制日志，直到它们过期或达到日志文件数的限制）。在这种情况下，日志会在 2 天后过期。

配置好后重启Mysql:

systemctl restart mysqld.service
  • Java环境检查
echo $JAVA_HOME
image-20250211111637904
2.新建测试库和表
 CREATE DATABASE IF NOT EXISTS canal default charset utf8 COLLATE utf8_general_ci;CREATE TABLE `test_user` (`id` bigint unsigned NOT NULL AUTO_INCREMENT,`name` varchar(200) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '姓名',`sex` varchar(20) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '性别',`tel` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '电话',PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
3.新建Es索引
curl -X PUT "http://<your es IP>:9200/test_user" -H 'Content-Type: application/json' -u <es账号>:<es 密码> -d'
{"mappings": {"properties": {"id": {"type": "long"},"title": {"type": "text"},"sex": {"type": "text"},"tel": {"type": "text"}}}
}
'

Canal_Server_109">二.部署 Canal Server

2.1 解压安装包
# 创建目录
mkdir -p /opt/canal/server /opt/canal/adapter# 解压 Server
tar -zxvf canal.deployer-1.1.8.tar.gz -C /opt/canal/server# 解压 Adapter
tar -zxvf canal.adapter-1.1.8.tar.gz -C /opt/canal/adapter
Canal_Server_124">2.2 配置 Canal Server

修改配置文件 /opt/canal/server/conf/canal.properties：

# tcp bind ip
canal.ip =127.0.0.1
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112# 目标实例名称（默认 example）
canal.destinations = example# 持久化模式（默认内存，可选 H2/MySQL）
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml

这里主要修改canal.ip其他保持默认即可。

修改实例配置 /opt/canal/server/conf/example/instance.properties：

#被同步的mysql地址，填写自己的IP地址
canal.instance.master.address=127.0.0.1:3306
#第一步创建的数据库从库权限账号/密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=Password@123
#数据库连接编码 
canal.instance.connectionCharset = UTF-8 
#Binlog 过滤规则（监控所有库表）
canal.instance.filter.regex=.*\\..*
#指定了 Canal 消费者（比如 MQ 客户端）读取和写入消息的目标主题，保持默认即可
canal.mq.topic=example
Canal_Server_161">2.3 启动 Canal Server
cd /opt/canal/server/bin
./startup.sh# 查看日志
tail -f /opt/canal/server/logs/canal/canal.log
tail -f /opt/canal/server/logs/example/example.log

image-20250211153418697

image-20250211153400835

image-20250211153538656

可以看到日志没有明显报错，且进程已经启动，则表示Canal Server已经启动成功。

image-20250211153842261

Canal_Adapter_Elasticsearch_182">三. 部署 Canal Adapter（同步到 Elasticsearch）

3.1 配置 Adapter

修改配置文件 /opt/canal/adapter/conf/application.yml：

server:port: 8081
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8default-property-inclusion: non_nullcanal.conf:mode: tcp # 客户端的模式，可选tcp kafka rocketMQflatMessage: true # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效zookeeperHosts:    # 对应集群模式下的zk地址syncBatchSize: 1000 # 每次同步的批数量retries: 0 # 重试次数, -1为无限重试timeout: # 同步超时时间, 单位毫秒accessKey:secretKey:consumerProperties:# canal tcp consumercanal.tcp.server.host: 127.0.0.1:11111 #配置canal-server的地址canal.tcp.zookeeper.hosts:canal.tcp.batch.size: 500canal.tcp.username:canal.tcp.password:srcDataSources: # 源数据库配置defaultDScanal是测试数据库url: jdbc:mysql://<yourIP>:3306/canal?useUnicode=true&useSSL=true #数据库连接,canal是测试用的数据库username: root #数据库账号password: Pass@1234 #数据库密码canalAdapters: # 适配器列表- instance: example # canal实例名,和上述Server的配置一样groups: # 分组列表- groupId: g1 # 分组id, 如果是MQ模式将用到该值outerAdapters:- name: logger # 日志打印适配器- name: es8 # ES同步适配器根据自己的es版本来hosts: <your IP>:9200 # ES连接地址properties:mode: rest # 模式可选transport端口(9300) 或者 rest端口(9200)security.auth: elastic:123456 #  连接es的用户和密码，仅rest模式使用cluster.name: elasticsearch # ES集群名称

如何获取es集群名称,命令输出的cluster_name就是上面需要配置的集群名字：

curl -u elastic:<esPass> -X GET "http://<es IP>:9200/_cluster/health?pretty"

image-20250211170653195

3.2 配置数据映射

创建 Elasticsearch 映射文件 /opt/canal/adapter/conf/es8/mytest_user.yml：

dataSourceKey: defaultDS # 源数据源的key, 对应上面application配置的srcDataSources中的值
destination: example  # canal的instance或者MQ的topic
groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:_index: test_user # es 的索引名称_id: _id  # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配sql: "SELECTtb.id AS _id,tb.name,tb.sex,tb.telFROMtest_user us"        # sql映射etlCondition: "where p.id>={}"   #etl的条件参数commitBatch: 3000   # 提交批大小
3.3 启动 Adapter
cd /opt/canal/adapter/bin
./startup.sh#查看日志
tail -f /opt/canal/adapter/logs/adapter/adapter.log

会输出很多数据库变更的日志：

image-20250211171145018

image-20250211171208031

4. 验证同步

4.1 插入测试数据到 MySQL
#执行sql
INSERT INTO test_user (name, sex, tel) VALUES ('Paco', '男', '123456789');

image-20250211171534121

image-20250211171547780

4.2 查询 Elasticsearch
curl -u elastic:<esPass> -X GET "http://<esIP>:9200/test_user/_search?pretty"

也可以在工具上查看，这边是Eage插件:

image-20250211171753695

image-20250211171808091

至此，即可验证可同步成功。我们可以修改数据测试，看是否能同步。

image-20250211171849802

image-20250211172502715

然后我们测试修改Es的数据:

image-20250211172542248

image-20250211172555087

可以发现数据库并没有变，至此Canal单向实时同步Mysql Binlog至Elasticsearch就配置完成了。


http://www.ppmy.cn/news/1571704.html

相关文章

全排列(力扣46)

全排列(力扣46)

这道题让我们求这个集合有多少种排列方式&#xff0c;那么与之前组合问题的不同就在于要考虑元素之间的顺序了&#xff0c;所以每一层递归的or循环的起始值无需变量控制&#xff0c;都从0开始。但是一个排列中不能出现相同元素&#xff0c;所以别忘了去重&#xff0c;这里的去重…
阅读更多...
在 Vue 项目中使用 SQLite 数据库的基础应用

在 Vue 项目中使用 SQLite 数据库的基础应用

目录 一、环境准备二、数据库连接与操作1. 创建数据库连接2. 创建表3. 插入数据4. 查询数据5. 更新数据6. 删除数据 三、在 Vue 组件中使用 SQLite 一、环境准备 安装 Node.js 和 npm&#xff1a;确保已安装 Node.js 和 npm。 创建 Vue 项目&#xff1a;使用 Vue CLI 创建一个…
阅读更多...
java进阶之并发编程一ReentrantLock的实际应用和线程中断EXAMPLE

java进阶之并发编程一ReentrantLock的实际应用和线程中断EXAMPLE

引言:继上一篇ReentrantLock的介绍来做俩个小demo。 实现3个线程分别打印指定数字和线程死锁进行线程中断。 上一篇:<<java进阶之并发编程一ReentrantLock同步锁的学习和syncthronized的区别>> **demo1:**ReentrantLock搭配三个线程分别打印指定的数字,直接上代…
阅读更多...
Android Studio设置增量build

Android Studio设置增量build

Android Studio设置build时只build新增的代码 而不是build全部代码&#xff0c;从而提高build速度
阅读更多...
疯狂前端面试题(二)

疯狂前端面试题(二)

一、Webpack的理解 Webpack 是一个现代 JavaScript 应用程序的静态模块打包工具。Webpack 能够将各种资源&#xff08;JavaScript、CSS、图片、字体等&#xff09;视为模块&#xff0c;并通过依赖关系图将这些模块打包成一个或多个最终的输出文件&#xff08;通常是一个或几个…
阅读更多...
Linux: 调整套接字缓冲区大小相关内核参数

Linux: 调整套接字缓冲区大小相关内核参数

Linux: 调整套接字缓冲区大小相关内核参数 内核参数关于套接字缓冲区大小相关的设置&#xff0c;这些参数控制了TCP和UDP套接字的接收和发送缓冲区的最大值、默认值以及动态调整范围。 当前配置 net.core.rmem_max 212992 # 最大接收缓冲区大小&#xff08;字节&#…
阅读更多...
一致性Hash算法延伸至Redis分片扩容使Lua脚本失效如何解决

一致性Hash算法延伸至Redis分片扩容使Lua脚本失效如何解决

文章部分内容来源&#xff1a;小林coding 问题场景&#xff1a;我们需要用Lua脚本&#xff0c;并且这个Lua脚本需要用到两个Key&#xff0c;但这两个Key必须命中同一台机器才可以&#xff0c;不然Lua脚本就会执行失败。如果集群扩容可能会导致两个Key落到不同的节点上导致Lua脚…
阅读更多...
linux 内核结构基础

linux 内核结构基础

linux 内核对象基础 1.linux 的 kobj(struct kobject) 类2.linux 中的 kset3. linux 下的 ktype (kobj_type 类)4. kobj 的使用理解 1.linux 的 kobj(struct kobject) 类 kobj 是 linux 下的高级抽象类定义&#xff0c;用于派生所有其余的类定义,比如设备类定义struct device.…
阅读更多...
最新文章

Copyright @ 2022~2023