使用canal實時同步MySQL資料到es

軟體版本資訊
安裝 elasticsearch
安裝 kibana
下載和安裝canal
- 1、下載canal
- 2、配置MySQL
- 3、配置 canal-server (canal-deploy)
- 4、配置 canal-adapter
- 5、配置canal-admin
同步MySQL到es
- 1、創建es索引
- 2、在MySQL資料庫中創建資料
- 3、修改MySQL資料
- 4、對MySQL資料進行洗掉操作
常規錯誤解決

軟體版本資訊

由于不同版本的MySQL、Elasticsearch和canal會有兼容性問題，所以我們先對其使用版本做個約定，

軟體	版本
作業系統	windows 10
elasticsearch	7.16.2
kibana	7.16.2
canal	1.1.5
MySQL	5.7
jdk	1.8.0

elasticsearch下載地址：https://www.elastic.co/cn/downloads/past-releases#elasticsearch
kibana下載地址：https://www.elastic.co/cn/downloads/past-releases#kibana
canal下載地址：https://github.com/alibaba/canal/releases

安裝 elasticsearch

根據自己所需下載對應 elasticsearch 版本，我這里使用的是 7.16.2 版本
下載地址：https://www.elastic.co/cn/downloads/past-releases#elasticsearch
下載完成后解壓 elasticsearch-7.16.2-windows-x86_64.zip 壓縮包到自己本地路徑上，解壓后目錄如下：

進入\elasticsearch-7.16.2\config 路徑，修改 elasticsearch.yml 檔案

#配置es集群名稱：es
cluster.name: es
#配置es埠：9200
http.port: 9200
#在檔案中末尾處添加下面引數，開啟x-pack驗證
xpack.security.enabled: true
xpack.license.self_generated.type: basic
xpack.security.transport.ssl.enabled: true
#關閉geoip資料庫的更新，要不然啟動es會報錯
ingest.geoip.downloader.enabled: false

設定 elasticsearch 賬號密碼
打開cmd命令符，進入目錄 \elasticsearch-7.16.2\bin ，執行設定用戶名和密碼的命令,這里需要為6個用戶分別設定密碼，elastic, apm_system, kibana_system, logstash_system, beats_system, remote_monitoring_user
```
#首先啟動elasticsearch，執行命令
elasticsearch.bat
#另外打開命令符進入 \bin 路徑，執行下面命令設定用戶密碼：123456
elasticsearch-setup-passwords interactive
```
下圖表示密碼設定成功，重啟 elasticsearch

測驗啟動是否成功
在瀏覽器輸入以下地址：http://127.0.0.1:9200/，輸入用戶：elastic，密碼：123456
看到以下內容表示啟動成功：

{
  "name" : "DESKTOP-M1CKQ7R",
  "cluster_name" : "es",
  "cluster_uuid" : "8gUkj244QD2gkTcrRNhbyw",
  "version" : {
    "number" : "7.16.2",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "2b937c44140b6559905130a8650c64dbd0879cfb",
    "build_date" : "2021-12-18T19:42:46.604893745Z",
    "build_snapshot" : false,
    "lucene_version" : "8.10.1",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

安裝 kibana

根據自己所需下載 kibana 版本，我這里使用的是 7.16.2 版本
下載地址：https://www.elastic.co/cn/downloads/past-releases#kibana
解壓 kibana-7.16.2-windows-x86_64.zip 壓縮包到本地目錄下：

進入路徑 \kibana-7.16.2\config 修改 kibana.yml 檔案

#打開kibana埠：5601
server.port: 5601
#設定kibana服務host：127.0.0.1
server.host: "127.0.0.1"
#配置kibana連接es的地址和埠
elasticsearch.hosts: ["http://127.0.0.1:9200"]
#配置kibana連接es的驗證用戶和密碼
elasticsearch.username: "elastic"
elasticsearch.password: "123456"
#設定中文編碼
i18n.locale: "zh-CN"
#在末尾處添加以下屬性，開啟es對kibana的xpack認證
xpack.monitoring.ui.container.elasticsearch.enabled: true

啟動 kibana
```
#打開cmd命令符，進入路徑\kibana-7.16.2\bin，輸出命令：
kibana.bat
```
瀏覽器訪問：http://127.0.0.1:5601/
如果進入以下界面，表示 kibana 啟動成功

輸入用戶：elastic，密碼：123456 登陸就可以進入使用 kbana 了

下載和安裝canal

1、下載canal

首先我們需要下載canal的各個組件canal-server、canal-adapter、canal-admin，大家可以根據自己需要下載版本，我這里使用的版本是1.1.5版本
下載地址：https://github.com/alibaba/canal/releases

2、配置MySQL

配置資料庫 my.ini 檔案

[mysqld]
#設定serveri_id
server_id=1
#開啟二進制日志功能
log-bin=mall-mysql-bin 
#設定使用的二進制日志格式（mixed,statement,row）
binlog_format=row

配置成功后重啟 MySQL ，查看 binlog 是否啟用：

mysql> show variables like '%log_bin%';
+---------------------------------+------------------------------------------+
| Variable_name                   | Value                                    |
+---------------------------------+------------------------------------------+
| log_bin                         | ON                                       |
| log_bin_basename                | D:\MySQL5.7.26\data\mall-mysql-bin       |
| log_bin_index                   | D:\MySQL5.7.26\data\mall-mysql-bin.index |
| log_bin_trust_function_creators | OFF                                      |
| log_bin_use_v1_row_events       | OFF                                      |
| sql_log_bin                     | ON                                       |
+---------------------------------+------------------------------------------+
6 rows in set, 1 warning (0.00 sec)

查看binlog模式：

mysql> show variables like '%binlog_format%';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW   |
+---------------+-------+
1 row in set, 1 warning (0.01 sec)

擁有從庫權限的賬號，用于訂閱binlog，這里創建的賬號為canal:root；

mysql> create user canal identified by 'root';
mysql> grant select, replication slave, replication client on *.* to 'canal'@'%';
mysql> flush privtleges;

創建一個用于測驗的資料庫：canal

mysql> CREATE DATABASE canal;
mysql> CREATE DATABASE IF NOT EXISTS canal default charset utf8 COLLATE utf8_general_ci;

然后創建一張資料庫表 product

mysql> use canal;
mysql> 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;

3、配置 canal-server (canal-deploy)

canal-server（canal-deploy）：可以直接監聽MySQL的binlog，把自己偽裝成MySQL的從庫，只負責接收資料，并不做處理，
解壓我們下載好的壓縮包 canal.deployer-1.1.5-SNAPSHOT.tar.gz到本地目錄，解壓完成后目錄結構如下：

├── bin
│   ├── restart.sh
│   ├── startup.bat
│   ├── startup.sh
│   └── stop.sh
├── conf
│   ├── canal_local.properties
│   ├── canal.properties
│   ├── example
│   │   └── instance.properties
│   ├── logback.xml
│   ├── metrics
│   │   └── Canal_instances_tmpl.json
│   └── spring
├── lib
├── logs
└── plugin

進入 conf 路徑，修改組態檔 canal.properties,主要修改

#canal的server地址：127.0.0.1
canal.ip =127.0.0.1
#canal埠
canal.port = 11111

進入 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=root
# 資料庫連接編碼
canal.instance.connectionCharset = UTF-8
# 需要訂閱binlog的表過濾正則運算式
canal.instance.filter.regex=.*\\..*

打開cmd命令符，進入 bin 目錄下，執行啟動命令：

startup.bat

啟動后，查看服務日志資訊： logs/canal/canal.log 在這里插入圖片描述
查看instance日志資訊：logs/example/example.log ，如下canal-server 已成功啟動

4、配置 canal-adapter

canal-adapter：相當于canal的客戶端，會從canal-server中獲取資料，然后對資料進行同步，可以同步到MySQL、Elasticsearch和HBase等存盤中去，
解壓我們下載好的壓縮包 canal.adapter-1.1.5.tar.gz 到本地目錄，解壓完成后目錄結構如下：

├── bin
│   ├── adapter.pid
│   ├── restart.sh
│   ├── startup.bat
│   ├── startup.sh
│   └── stop.sh
├── conf
│   ├── application.yml
│   ├── es6
│   ├── es7
│   │   ├── biz_order.yml
│   │   ├── customer.yml
│   │   └── mytest_user.yml
│   ├── hbase
│   ├── kudu
│   ├── logback.xml
│   ├── META-INF
│   │   └── spring.factories
│   └── rdb
├── lib
├── logs
│   └── adapter
│       └── adapter.log
└── plugin

修改組態檔conf/application.ym

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 consumer
    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://127.0.0.1:3306/canal?useUnicode=true&useSSL=true
      username: canal
      password: root
  canalAdapters: # 配接器串列
  - instance: example # canal實體名或者MQ topic名
    groups: # 分組串列
    - groupId: g1 # 分組id, 如果是MQ模式將用到該值
      outerAdapters:
      - name: logger # 日志列印配接器
      - name: es7 # ES同步配接器
        hosts: http://127.0.0.1:9200 # ES連接地址
        properties:
          mode: rest # 模式可選transport(9300) 或者 rest(9200)
          security.auth: elastic:123456 #  連接es的用戶和密碼，僅rest模式使用
          cluster.name: es # ES集群名稱, 與es目錄下 elasticsearch.yml檔案cluster.name對應

修改 canal-adapter/conf/es7/mytest_user.yml 檔案，用于配置MySQL中的表與Elasticsearch中索引的映射關系

dataSourceKey: defaultDS # 源資料源的key, 對應上面配置的srcDataSources中的值
destination: example  # canal的instance或者MQ的topic
groupId: g1 # 對應MQ模式下的groupId, 只會同步對應groupId的資料
esMapping:
  _index: canal_product # es 的索引名稱
  _id: _id  # es 的_id, 如果不配置該項必須配置下面的pk項_id則會由es自動分配
  sql: "SELECT
         p.id AS _id,
         p.title,
         p.sub_title,
         p.price,
         p.pic
        FROM
         product p"        # sql映射
  etlCondition: "where p.id>={}"   #etl的條件引數
  commitBatch: 3000   # 提交批大小

啟動 canal-adapter

#打開cmd命令符，bin路徑，執行命令
startup.bat

以下圖片表示啟動成功：
啟動canal

5、配置canal-admin

canal-admin：為canal提供整體配置管理、節點運維等面向運維的功能，提供相對友好的WebUI操作界面，方便更多用戶快速和安全的操作，
解壓我們下載好的壓縮包 canal.admin-1.1.5.tar.gz 到本地目錄，解壓完成后目錄結構如下：

├── bin
│   ├── restart.sh
│   ├── startup.bat
│   ├── startup.sh
│   └── stop.sh
├── conf
│   ├── application.yml
│   ├── canal_manager.sql
│   ├── canal-template.properties
│   ├── instance-template.properties
│   ├── logback.xml
│   └── public
│       ├── avatar.gif
│       ├── index.html
│       ├── logo.png
│       └── static
├── lib
└── logs

創建 canal-admin 需要使用的資料庫canal_manager，創建SQL腳本為/mydata/canal-admin/conf/canal_manager.sql，會創建如下表；
在這里插入圖片描述修改組態檔conf/application.yml，按如下配置即可，主要是修改資料源配置和canal-admin的管理賬號配置，注意需要用一個有讀寫權限的資料庫賬號，比如管理賬號root:root；

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
 
spring.datasource:
  address: 127.0.0.1:3306
  database: canal_manager
  username: root  #資料庫管理員賬號
  password: root  #資料庫管理員賬號密碼
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1
 
canal:
  adminUser: admin
  adminPasswd: admin

啟動canal-admin

bin/startup.bat

訪問canal-admin的Web界面，輸入賬號密碼admin:123456即可登錄，訪問地址：http://127.0.0.1:8089
在這里插入圖片描述

同步MySQL到es

1、創建es索引

瀏覽器訪問：http://127.0.0.1:5601/，用戶：elastic，密碼：123456
在這里插入圖片描述

# 在kibana工具里創建es索引
PUT canal_product
{
  "mappings": {
    "properties": {
      "title": {
        "type": "text"
      },
      "sub_title": {
        "type": "text"
      },
      "pic": {
        "type": "text"
      },
      "price": {
        "type": "double"
      }
    }
  }
}

在這里插入圖片描述

2、在MySQL資料庫中創建資料

# 之后使用如下SQL陳述句在資料庫中創建一條記錄 
INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 5, '小米8', ' 全面屏游戲智能手機 6GB+64GB', 1999.00, NULL );

創建成功后，在Elasticsearch中搜索，發現資料已經同步了；

GET canal_product/_search

在這里插入圖片描述

3、修改MySQL資料

# 再使用如下SQL對資料進行修改；
UPDATE product SET title='小米10' WHERE id=5

在這里插入圖片描述

4、對MySQL資料進行洗掉操作

# 再使用如下SQL對資料進行洗掉操作；
DELETE FROM product WHERE id=5

在這里插入圖片描述

參考鏈接：https://blog.csdn.net/zhenghongcs/article/details/109476376

常規錯誤解決

canal-adapter 啟動報錯：

2022-02-16 14:27:11.043 [Thread-2] ERROR c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: es7 failed
java.lang.IllegalStateException: Extension instance(name: es, class: interface com.alibaba.otter.canal.client.adapter.OuterAdapter)  could not be instantiated: class could not be found

這是由于 druid 包沖突導致的

問題解決：

下載原始碼包
下載地址為：https://github.com/alibaba/canal/archive/refs/tags/canal-1.1.5.tar.gz

解壓后，使用IDEA打開，定位到 client-adapter.escore 模塊，將 pom.xml 的 druid 更新為：

<dependency>
	<groupId>com.alibaba</groupId>
	<artifactId>druid</artifactId>
	<scope>provided</scope>
</dependency>

更新后，在專案根目錄下執行命令
```
mvn clean package
```
然后到 canal-canal-1.1.5/client-adapter/es7x/target 下將打包好的 client-adapter.es7x-1.1.5-jar-with-dependencies.jar 替換掉 canal-adapter/plugin 下原來的 jar包，然后重新啟動 canal-adapter

參考鏈接：https://ask.csdn.net/questions/7501103

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/432149.html

標籤：其他

上一篇：R語言匯入資料檔案（資料匯入、加載、讀取）、使用readr包中的read_csv函式匯入逗號分割檔案CSV（Comma Delimited Text File）

下一篇：Flink查詢關聯Hbase輸出