利用canal进行MySQL到ES的数据实时同步

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. 环境信息

本文以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往往部署在不同的网络区域，这里就涉及到网络安全访问策略的问题，下面梳理出了需要的网络安全访问策略。

代码007普通

打赏 收藏 海报 链接