基于Canal的MySQL=＞ES数据同步方案
文章目錄
- 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？相信不了解這兩個數據庫的區別的同學很容易有這樣的疑問。下面我來解答這個問題。
首先兩款數據庫在功能上的區別：
| 關系型 | 搜索引擎 |
| 單體 | 分布式 |
| OLTP | OLAP |
| 支持事務 | 不支持事務 |
| SQL | DSL |
1.2 性能指標
再來看一下兩種數據庫的性能對比
| 檢索性能 | ★★★ | ★★★★★ |
| 擴展能力 | ★★ | ★★★★★ |
| 寫入實時性 | ★★★★ | ★★（可配置） |
| 準確性 | ★★★★ | ★★ |
| 支持的數據類型 | ★★★ | ★★★★ |
| 靈活性 | ★★★ | ★★★★ |
| 事務支持 | ★★★ | ★★★★ |
可以看到，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服務
驗證是否開啟成功
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字段。
服務啟動
注意：
- 索引msb_order的mapping必須提前創建好
- 索引中的id字段是_id，因此需要查詢的時候需要id as _id
總結
以上是生活随笔為你收集整理的基于Canal的MySQL=＞ES数据同步方案的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： C++实现传输文件
- 下一篇： Git/Bitbucket Workfl