canal同步mysql数据到es中
maven依赖
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.4</version>
</dependency>
开启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)
mysql用户数据准备
创建一个canal用户并对其进行授权
CREATE USER canal IDENTIFIED BY 'Cjz123456.';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
创建一个测试用户数据库及表
create database canal;<br>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;
创建存放canal的目录
mkdir /usr/local/canal-server
mkdir /usr/local/canal-adpter
mkdir /usr/local/canal-admin
部署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
部署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中索引的映射关系
最后更新于 2026-02-18 18:15