1. 背景
项目中业务数据量比较大，每类业务表都达到千万级别，虽然做了分库分表，每张表数据控制在300W以下，但是效率还是达不到要求，为了提高查询效率，打算使用ES进行数据查询。
2. 同步原理
canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送 dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)，转换为json格式 Canal 客户端通过 TCP 协议或 MQ 形式监听 Canal 服务端，同步数据到 ES。
优点： 可以完全和业务代码解耦，增量日志订阅。
缺点：实时性不高，订阅mysql日志，DB中数据事务成功后，开始同步至canal。
3. 环境信息
名称版本MySQL8.0.35elasticsearch7.17.9canal1.1.7jdk1.8
本文以MySQL中的表ibds2.proof_data_history_202311为例进行数据同步实验。
4. 部署配置
本文的的重点是讲述canal的配置与使用，不再详细描述MySQL与ES的部署和配置。
4.1. MySQL
MySQL的部署与配置相对较简单，需要主要的是需要开启binlog，且配置binlog-format 为ROW模式。此处不再做详细描述。
4.2. Elasticsearch
ES的部署与配置也比较简单，此处不再做详细描述。
建议安装同版本的分词器插件。
4.3. canal
版本下载地址：
deployer：
https://download.fastgit.org/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz
adapter：
https://download.fastgit.org/alibaba/canal/releases/download/canal-1.1.7/canal.adapter-1.1.7.tar.gz
4.3.1. deployer
4.3.1.1. 配置
将canal.deployer-1.1.7.tar.gz上传到/data/canal/Server，并进行解压：
cd /data/canal/Server
tar -zxvf canal.deployer-1.1.7.tar.gz
修改配置文件conf/example/instance.properties，按如下配置即可，主要是修改数据库相关配置：
## mysql serverId , v1.0.26+ will autoGen 不要与MySQL的id重复
canal.instance.mysql.slaveId=10
# enable gtid use true/false 是否使用gtid
canal.instance.gtidon=true
# position info MySQL数据库信息
canal.instance.master.address=192.168.10.101:3307
# username/password
canal.instance.dbUsername=root
canal.instance.dbPassword=X87_5w2Anxp3
canal.instance.connectionCharset = UTF-8
# table regex 需要订阅binlog的表的过滤正则表达式
canal.instance.filter.regex=.*\\..*
4.3.1.2. 启动
进入bin目录，启动deployer：
cd /data/canal/Server/bin
./startup.sh
4.3.2. adapter
在adapter的配置中，一定要先在ES中创建索引，然后在启动adapter。
4.3.2.1. 配置
将canal.adapter-1.1.7.tar.gz上传到/data/canal/Adapter，并进行解压：
cd /data/canal/Adapter
tar -zxvf canal.adapter-1.1.7.tar.gz
修改配置文件conf/application.yml，按如下配置即可，主要是修改canal-server配置、数据源配置和客户端适配器配置：
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 rabbitMQ
flatMessage: true
zookeeperHosts:
syncBatchSize: 1000
retries: -1
timeout:
accessKey:
secretKey:
consumerProperties:
# canal tcp consumer
canal.tcp.server.host: 192.168.10.101:11111 # 需要修改
canal.tcp.zookeeper.hosts:
canal.tcp.batch.size: 500
canal.tcp.username:
canal.tcp.password:
# kafka consumer
kafka.bootstrap.servers: 127.0.0.1:9092
kafka.enable.auto.commit: false
kafka.auto.commit.interval.ms: 1000
kafka.auto.offset.reset: latest
kafka.request.timeout.ms: 40000
kafka.session.timeout.ms: 30000
kafka.isolation.level: read_committed
kafka.max.poll.records: 1000
# rocketMQ consumer
rocketmq.namespace:
rocketmq.namesrv.addr: 127.0.0.1:9876
rocketmq.batch.size: 1000
rocketmq.enable.message.trace: false
rocketmq.customized.trace.topic:
rocketmq.access.channel:
rocketmq.subscribe.filter:
# rabbitMQ consumer
rabbitmq.host:
rabbitmq.virtual.host:
rabbitmq.username:
rabbitmq.password:
rabbitmq.resource.ownerId:
srcDataSources:
defaultDS:
url: jdbc:mysql://192.168.10.101:3307/ibds2?useUnicode=true # 需要修改
username: root # 需要修改
password: X87_5w2Anxp3 # 需要修改
canalAdapters:
- instance: example # canal instance Name or mq topic name
groups:
- groupId: g1
outerAdapters:
- name: logger
# - name: rdb
# key: mysql1
# properties:
# jdbc.driverClassName: com.mysql.jdbc.Driver
# jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
# jdbc.username: root
# jdbc.password: 121212
# druid.stat.enable: false
# druid.stat.slowSqlMillis: 1000
# - name: rdb
# key: oracle1
# properties:
# jdbc.driverClassName: oracle.jdbc.OracleDriver
# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
# jdbc.username: mytest
# jdbc.password: m121212
# - name: rdb
# key: postgres1
# properties:
# jdbc.driverClassName: org.postgresql.Driver
# jdbc.url: jdbc:postgresql://localhost:5432/postgres
# jdbc.username: postgres
# jdbc.password: 121212
# threads: 1
# commitSize: 3000
# - name: hbase
# properties:
# hbase.zookeeper.quorum: 127.0.0.1
# hbase.zookeeper.property.clientPort: 2181
# zookeeper.znode.parent: /hbase
- name: es7 # 需要修改
hosts: 192.168.10.101:9300 # 127.0.0.1:9200 for rest mode # 需要修改
properties:
mode: transport # or rest
# security.auth: test:123456 # only used for rest mode
cluster.name: my-es # 需要修改
# - name: kudu
# key: kudu
# properties:
# kudu.master.address: 127.0.0.1 # ',' split multi address
# - name: phoenix
# key: phoenix
# properties:
# jdbc.driverClassName: org.apache.phoenix.jdbc.PhoenixDriver
# jdbc.url: jdbc:phoenix:127.0.0.1:2181:/hbase/db
# jdbc.username:
# jdbc.password:
配置重点
jdbc:mysql://192.168.10.101:3307/ibds2?useUnicode=true中的数据库名称 cluster.name：es集群名称根据自己的实际的配置，我的是my-es -name: es7 这个很重要一会儿要用
添加配置文件canal-adapter/conf/es7/proof_202311.yml，用于配置MySQL中的表与Elasticsearch中索引的映射关系：
dataSourceKey: defaultDS # 与application.yml中的srcDataSources对应
destination: example # canal的instance
groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:
_index: proof_data_history_202311 # 要在es中创建的索引名称
_type: _doc
_id: id # 与sql中的id一致
upsert: true
sql: "SELECT
t.id as id,
t.biz_id as biz_id,
t.biz_type as biz_type,
t.encrypt_mode as encrypt_mode,
t.sign_user as sign_user,
t.tx_hash as tx_hash,
t.from_addr as from_addr,
t.data as data,
t.channel_id as channel_id,
t.block_height as block_height,
t.tx_time as tx_time
FROM
proof_data_history_202311 t"
commitBatch: 3000
4.3.2.2. 创建索引
根据MySQL的表结构信息，在ES中创建对应的索引信息。
在ES中创建索引的方法有多种，推荐两种方法：
kinbana：
PUT /proof_data_history_202311
{
"mappings":{
"properties":{
"id": {
"type": "long"
},
"biz_id": {
"type": "text"
},
"biz_type": {
"type": "long"
},
"encrypt_mode": {
"type": "long"
},
"sign_user": {
"type": "text"
},
"tx_hash": {
"type": "text"
},
"from_addr": {
"type": "text"
},
"data": {
"type": "text"
},
"channel_id": {
"type": "text"
},
"block_height": {
"type": "long"
},
"tx_time": {
"type": "date"
}
}
}
}
命令行：
curl -XPUT "http://192.168.10.101:9200/proof_data_history_202311" -H 'Content-Type: application/json' -d'
{
"mappings":{
"properties":{
"id": {
"type": "long"
},
"biz_id": {
"type": "text"
},
"biz_type": {
"type": "long"
},
"encrypt_mode": {
"type": "long"
},
"sign_user": {
"type": "text"
},
"tx_hash": {
"type": "text"
},
"from_addr": {
"type": "text"
},
"data": {
"type": "text"
},
"channel_id": {
"type": "text"
},
"block_height": {
"type": "long"
},
"tx_time": {
"type": "date"
}
}
}
}'
4.3.2.3. 启动adapter
cd /data/canal/Adapter/bin
./startup.sh
4.3.3. 数据初始化
如果mysql中有数据就需要调用一次全量同步，如果mysql没数据，或者数据没用，就不需要调用此步骤。
canal-adapter提供一个REST接口可全量同步数据到ES：
# 全量同步proof_data_history_202311表数据到es的proof_data_history_202311索引中，路径中的es7是上一步的文件夹名
curl -X POST http://192.168.10.101:8081/etl/es7/proof_202311.yml
以120万行实验数据为例，分别进行ES中没有数据和已有数据的两种不同情况下的全量同步实验。
ES中没有数据，同步时间86秒：
2023年 11月 28日 星期二 09:17:55 CST开始执行全量同步
{"succeeded":true,"resultMessage":"导入ES 数据：1200013 条"}
2023年 11月 28日 星期二 09:19:21 CST完成执行全量同步
ES中已有数据，同步时间87秒：
2023年 11月 28日 星期二 09:27:46 CST开始执行全量同步
{"succeeded":true,"resultMessage":"导入ES 数据：1200013 条"}
2023年 11月 28日 星期二 09:29:03 CST完成执行全量同步
5. 补充资料
在实际生产中，server和adapter往往部署在不同的网络区域，这里就涉及到网络安全访问策略的问题，下面梳理出了需要的网络安全访问策略。
源端目标端Adapter端IP数据源端MySQL地址和端口Adapter端IP数据目的地ES地址和端口（9200、9300）Adapter端IPServer端地址和11110、11111、11112端口Server端IP数据源端MySQL地址和端口
精彩内容
您阅读本篇文章共花了： 大家都在找：
Mysql：mysql主从复制
elasticsearch：elasticsearch怎么读
发表评论