文章目錄

• 1、MySQL和ES的主要區別？
• • 1.1 功能性
  • 1.2 性能指標
  • 1.3 在搜索業務上的區別
  • • 1.3.1 查詢
    • 1.3.2 檢索
• 2、為什么要做數據同步
• • 2.1 檢索性能
  • 2.2 寫入性能
  • 2.3 事務支持
• 3、Canal
• • 3.1 Canal是啥
  • 3.2 應用場景
  • 3.3 原理機制
  • • 3.3.1 MySQL主備復制原理
    • 3.3.2 canal 工作原理
  • 3.3 優勢
• 4 基于canal的MySQL=>ES數據同步方案
• • 4.1 環境：
  • 4.2 下載地址：
  • 4.3 步驟
  • • 4.3.1 保證`Elasticsearch`服務可用
    • 4.3.2 保證`MySQL`服務可用
    • 4.3.3 開啟MySQL的binary log（主備模式）
    • 4.4.4 canal-deployer
    • 4.4.5 canal-admin
  • 4.4.6 canal-adapter

1、MySQL和ES的主要區別？

1.1 功能性

MySQL作為最常用的DB之一，在DB-Ranking排名常年保持前三，僅次于Oracle。為什么還需要把數據同步至排名只有第八的Elasticsearch？相信不了解這兩個數據庫的區別的同學很容易有這樣的疑問。下面我來解答這個問題。

首先兩款數據庫在功能上的區別：

MySQLElasticsearch

關系型	搜索引擎
單體	分布式
OLTP	OLAP
支持事務	不支持事務
SQL	DSL

1.2 性能指標

再來看一下兩種數據庫的性能對比

MySQLElasticsearch

檢索性能	★★★	★★★★★
擴展能力	★★	★★★★★
寫入實時性	★★★★	★★（可配置）
準確性	★★★★	★★
支持的數據類型	★★★	★★★★
靈活性	★★★	★★★★
事務支持	★★★	★★★★

可以看到，Elasticsearch在檢索和擴展能力方面碾壓MySQL。實際上MySQL在數據量達到 2-3千萬的時候查詢性能開始顯著下降，而ES的檢索性能，實測在數據量1億左右，單節點1G內存，無任何優化的前提下仍能達到10ms以內的檢索性能。并且ES具有MySQL無法相比的橫向擴展能力，具備極高的可用性和擴展性。

1.3 在搜索業務上的區別

同樣是檢索結果，ES中的“檢索”和關系型數據庫的“查詢”是兩個不同的概念。搜索引擎的概念里，“檢索”和相關度緊密耦合。

1.3.1 查詢

關系數據庫中“查詢”的概念具有“完全相關性”，即搜索結果要么完全滿足要求，要么不滿足，而不存在“部分滿足”的概念。

1.3.2 檢索

而以ES為代表的搜索引擎中的“檢索”（泛指全文檢索），具備相關度的概念，即和搜索詞預期搜索結果相符的程度。影響相關度的因素有很多，搜索量，數據本身等都可能影響最終結果。ES以相關度評分來衡量相關度的結果，這里不什么是相關度評分，如果感興趣，可以異步“相關度評分”一文。

舉個簡單的例子：當用戶搜索詞為“小米手機”的時候，“查詢”的執行邏輯為field_name='小米手機'或者field_name like '%小米手機%'，即搜索結果的匹配邏輯為完全匹配或包含關系。而如果是檢索，假設分詞結果為“小米”和“手機”兩個詞項，搜索結果可能只包含“小米”或者只“手機”，如果設置了同義詞等邏輯，匹配出手表、華為也是符合結果的。即搜索結果可能完全不包含搜索詞。

2、為什么要做數據同步

MySQL是關系型數據庫，而Elasticsearch是搜索引擎的核心組件之一，屬于非關系型數據庫。有一句經典名言叫：存在即合理，兩款數據庫既然同時存在，那么他們必然有對方不可取代的地方。

這里只討論兩者最主要的不可替代特性，至于擴展能力、數據類型等問題本文暫不討論

2.1 檢索性能

ES擅長海量數據的檢索，支持PB級數據的秒級查詢，以億為單位的數據在ES看來只是起步而已，并且搜索結果具有非完全相關性，而是全文檢索。其相關度有專門的算法來計算，主要是TF-IDF和BM25，兩種算法這里不贅述，我的文章里有詳細介紹。而MySQL雖然也有“全文索引”，但是非常積累，不管存存儲效率還是檢索效率都遠不如ES，這也是其底層的數據結構和算法導致。普通索引目前底層數據結構為B+Trees，這種數據結構不適合存儲長文本類型的索引，會導致樹的深度過大，檢索效率極低。其底層原理我有單獨的文章詳細介紹（待更新）。

2.2 寫入性能

ES是OLAP系統，側重于海量數據的檢索，而寫入實時性并不是很高，默認1秒，也就是ES緩沖區Buffer的刷新間隔時間，不了解Elasticsearch寫入原理的同學可以暫時忽略。ES并非忽略了對寫入性能的優化，而是“有意為之”，其原因就在于基于ES的寫入機制，其寫入實時性和大數據檢索性能是一個二選一的行為。實際上生產環境中我們經常通過“犧牲寫入實時性”的操作來換取更高更快的“數據檢索”性能。

2.3 事務支持

正因為ES的寫入實時性并不高，如果我們需要快速響應用戶請求，我們常采取的手段就是使用緩存，但是在很多高并發的場景下，我們需要數據保持強一致性（如銀行系統），因此需要使用具有ACID特性的數據庫來支持，而MySQL就是一個比較好的選擇。

但是如果單單只用MySQL，又無法解決海量數據的檢索問題。其實在實際工作中常見的做法就是兩者結合起來，通過數據同步的操作將MySQL的實時數據同步至Elasticsearch，兩者各司其職互不干擾。

3、Canal

3.1 Canal是啥

canal [k?’n?l]，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費

早期阿里巴巴因為杭州和美國雙機房部署，存在跨機房同步的業務需求，實現方式主要是基于業務 trigger 獲取增量變更。從 2010 年開始，業務逐步嘗試數據庫日志解析獲取增量變更進行同步，由此衍生出了大量的數據庫增量訂閱和消費業務。

3.2 應用場景

Elasticsearch 不支持事務。 ES通常在分布式系統架構中承擔“搜索引擎”的角色，一般來說解決詞類問題，可以把ES和支持ACID特性的關系型數據庫結合起來使用。首先把對數據的更（增刪改）操作在RDB中執行，然后把這些動作同步到Elasticsearch。 通過這種方式，你將受益于數據庫 ACID 事務支持，并且在 Elasticsearch 中以正確的順序產生變更。 并發在關系數據庫中得到了處理。

以MySQL為例，如果要把數據從同步至ES，canal + binary log就是常用的一種增量解決方案。

3.3 原理機制

3.3.1 MySQL主備復制原理

• MySQL master 將數據變更寫入二進制日志( binary log, 其中記錄叫做二進制日志事件binary log events，可以通過 show binlog events 進行查看)
• MySQL slave 將 master 的 binary log events 拷貝到它的中繼日志(relay log)
• MySQL slave 重放 relay log 中事件，將數據變更反映它自己的數據

3.3.2 canal 工作原理

• canal 模擬 MySQL slave 的交互協議，偽裝自己為 MySQL slave ，向 MySQL master 發送dump 協議
• MySQL master 收到 dump 請求，開始推送 binary log 給 slave (即 canal )
• canal 解析 binary log 對象(原始為 byte 流)
  

3.3 優勢

• 準實時性
• 性能好
• 一勞永逸

4 基于canal的MySQL=>ES數據同步方案

4.1 環境：

Java和ES兼容性：https://www.elastic.co/cn/support/matrix#matrix_jvm

• JDK：1.8
• Elasticsearch：7.x
• MySQL： 5.7
• Canal： 1.1.4

4.2 下載地址：

• Github： https://github.com/alibaba/canal/Github
• 咻咻咻： https://github.com/fhefh2015/Fast-GitHub

4.3 步驟

4.3.1 保證Elasticsearch服務可用

使用cancal向ES同步數據之前需要保證ES的服務是可以正常訪問的

4.3.2 保證MySQL服務可用

不再贅述

4.3.3 開啟MySQL的binary log（主備模式）

配置MySQL：

server_id = 1 #開啟主從模式后每個MySQL節點的id log-bin = mysql-bin #bin-log的存儲位置 binlog-format = ROW #選擇存儲binlog日志方式為ROW模式

重啟MySQL服務
驗證是否開啟成功

SHOW VARIABLES LIKE 'log_bin'; log_bin ON #開啟

4.4.4 canal-deployer

配置：conf/example/instance.properties

#canal示例的slaveId canal.instance.mysql.slaveId=1234 #mysql地址 canal.instance.master.address= 127.0.0.1:3306 #用戶名 canal.instance.dbUsername = root #密碼 canal.instance.dbPassword = 123456 #指定需要同步的數據庫 canal.instance.defaultDatabaseName = msb_order #指定編碼方式 canal.instance.connectionCharset = UTF-8 #監控的是所有數據庫，所有的表改動都會監控到，這樣可能會浪費不少性能，可能我只想監控的是某一個數據庫下的表。 # .*\\..*表示監控所有數據庫，canal\\..*表示監控canal數據庫 canal.instance.filter.regex = .\*\\\\..\*

啟動： ./startup.sh（Linux）
驗證： demo

4.4.5 canal-admin

配置：conf/application.yml

server:port: 8089 spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8spring.datasource:address: 127.0.0.1:3306database: canal_managerusername: rootpassword: 123456driver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=falsehikari:maximum-pool-size: 30minimum-idle: 1 canal:adminUser: adminadminPasswd: admin

啟動管理服務
訪問服務：server_ip:8089

4.4.6 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 rocketMQ rabbitMQflatMessage: truezookeeperHosts:syncBatchSize: 1000retries: 0timeout:accessKey:secretKey:consumerProperties:# canal tcp consumercanal.tcp.server.host: 127.0.0.1:11111canal.tcp.zookeeper.hosts:canal.tcp.batch.size: 500canal.tcp.username:canal.tcp.password:# kafka consumerkafka.bootstrap.servers: 127.0.0.1:9092kafka.enable.auto.commit: falsekafka.auto.commit.interval.ms: 1000kafka.auto.offset.reset: latestkafka.request.timeout.ms: 40000kafka.session.timeout.ms: 30000kafka.isolation.level: read_committedkafka.max.poll.records: 1000# rocketMQ consumerrocketmq.namespace:rocketmq.namesrv.addr: 127.0.0.1:9876rocketmq.batch.size: 1000rocketmq.enable.message.trace: falserocketmq.customized.trace.topic:rocketmq.access.channel:rocketmq.subscribe.filter:# rabbitMQ consumerrabbitmq.host:rabbitmq.virtual.host:rabbitmq.username:rabbitmq.password:rabbitmq.resource.ownerId:srcDataSources:defaultDS:url: jdbc:mysql://127.0.0.1:3306/msb_order?useUnicode=trueusername: rootpassword: 123456canalAdapters:- instance: example # canal instance Name or mq topic namegroups:- groupId: g1outerAdapters:- name: logger- name: es7hosts: 127.0.0.1:9200 # 127.0.0.1:9200 for rest modekey: exampleKeyproperties:mode: rest # transport or rest# security.auth: test:123456 # only used for rest modecluster.name: elasticsearch

配置：conf/es7/my_order.yml
這里要注意，在寫SQL語句的時候，要保證a.id as _id,即把id屬性改為_id，以保證能正常寫入ES的_id字段。

dataSourceKey: defaultDS # 源數據源的key, 對應上面配置的srcDataSources中的值 outerAdapterKey: exampleKey # 對應application.yml中es配置的key destination: example # cannal的instance或者MQ的topic groupId: g1 # 對應MQ模式下的groupId, 只會同步對應groupId的數據_search esMapping:_index: msb_order # es 的索引名稱_type: _doc # es7 固定'_doc'，es8刪除_id: _id # es 的_id, 如果不配置該項必須配置下面的pk項_id則會由es自動分配upsert: truesql: "SELECT a.id as _id, a.customer_phone, a.customer_name, a.customer_region, a.customer_city, a.customer_district, a.customer_addr, a.failure_phenomenon, a.failure_phenomenon_text, a.book_date, a.service_required, a.buyshop_detail, a.buy_shop, a.buy_way, a.product_type, a.model_no, a.process_type, a.ex_order_no, a.customer_country, a.source, a.matnr, a.buyDate, a.serviceReq, a.create_time, a.result, a.rvstatus, a.ip_address FROM msb_order_1 a"commitBatch: 3000

服務啟動
注意：

• 索引msb_order的mapping必須提前創建好
• 索引中的id字段是_id，因此需要查詢的時候需要id as _id

總結

以上是生活随笔為你收集整理的基于Canal的MySQL=＞ES数据同步方案的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。