DB 與 Elasticsearch 混合應用之數據實時同步

作者介紹

李猛，Elastic Stack 深度用戶，通過 Elastic 工程師認證，2012年接觸 Elasticsearch，對 Elastic Stack 技術棧開發、架構、運維等方面有深入體驗，實踐過多種大中型項目；為企業提供 Elastic Stack 諮詢培訓以及調優實施；多年實戰經驗，愛搗騰各種技術產品，擅長大數據，機器學習，系統架構。

序言

前一篇文章《DB與ES混合之應用系統場景分析探討》，我們主要探討了混合場景下的多種模型映射類型，基本覆蓋了應用業務系統如何藉助Elasticsearch 來解決DB侷限性。

下面這篇文章，我們主要解決 DB 到 Elasticsearch 數據實時同步問題。

背景需求

DB與ES本質上是屬於不同應用領域的數據庫產品，混合應用在一起主要面臨2個問題：

1、同步實時性，數據在DB更新之後，需要多久才能更新到 Elasticsearch，多久的時間是應用系統可以接受的範圍，一般需要控制在1s以內，如果是分鐘以上，那這就屬於離線同步。

2、數據一致性，數據頻繁在DB變更修改，更新到Elasticsearch之後如何保證數據與DB一致，在容許的時間範圍內應用系統查詢的數據有效的，可接受的，如果變更出現覆蓋等，那數據是無效的，應用系統是不可接受的。

同步模式

在數據同步方面，主要有3種同步模式

1、推送模式，數據源將變更數據推送到目標源，如RabbitMQ產品，服務端會主動MQ發送到客戶端。

2、拉取模式，目標源定時去數據源拉取變更數據，如Mysql數據庫的數據主從同步機制，Slave會去Master拉取變更數據。

3、推拉結合，數據源與目標源之間，既有推送方式，也有拉取方式，此種模式一般會藉助於中間媒介實現，如基於Kafka產品的日誌應用，數據源（採集端）會將日誌數據發送到Kafka集群，目標源會定期的從Kafka拉取數據更新。

技術方案

從技術層面看，DB同步到ES有好多種方式

1、同步雙寫，更新DB時同步更新ES。此技術方案最簡單，附帶問題最多，數據衝突，數據覆蓋，數據丟失，處處是坑，謹慎選擇。

2、異步雙寫，更新DB之後，記錄一條MQ，MQ通知消費端，消費端反向查詢DB數據，最後更新到ES。此技術方案與業務系統耦合嚴重，需要更加業務需求獨立編寫，且每個業務都需要專門編寫相關程序，非常不利於快速響應需求。

3、CDC，全稱Change Data Capture，變更數據捕捉，從數據庫內部捕捉變更數據，將變更數據推送到中間程序中，中間程序邏輯實現同步推送到ES。CDC機制速度極快，數據精準，且與應用程序耦合少，可抽象脫離業務系統，適合大規模使用。如圖：

CDC機制原有設計是為了同類型數據庫之間數據同步，應用在主從同步高可用方面，所以同類型數據庫之間數據同步非常容易實現，數據庫廠商本身天然支持，經過多年實戰驗證高效可靠。相反，在異構數據庫之間實現數據同步是比較複雜的，數據鏈路長，中間涉及到的技術點特別多，且每一步都非常關鍵，下面就以本人所在公司採用的技術棧講述如何實現，以及一些技術關鍵點。

案例：MySQL 同步到 Elasticsearch

數據從Mysql同步到Elasticsearch主要涉及到以下幾個技術關鍵點
• Binlog機制
• Canal中間件
• Kakfa中間件
• 同步應用程序（業務型開發成中間件）

Binlog機制

Binlog是Mysql自帶功能機制，設計之初是為了數據庫之間主從同步
• Master主庫，啟動Binlog機制，將變更數據寫入Binlog文件
• Slave從庫，從Master主庫拉取binlon數據，回放Binlog，更新從庫數據
• 啟用Binlog注意以下幾點
• Master主庫一般會有多臺Slave訂閱，且Master主庫要支持業務系統實時變更操作，服務器資源會有瓶頸
• 需要同步的數據表一定要有主鍵

Canal中間件

Canal是Mysql的中間件產品，專門應用在數據同步
• 偽裝Slave從庫，
• 拉取Binlog數據，
• 回放Binlog數據，
• 解析Binlog數據為Json，報文記錄了新舊數據，數據庫數據表，更新方式

• 輸出數據，並保證變更順序，輸出目的源支持很多，常規的一般輸出到kafka
• 配置cannl注意以下幾點
• Canal基於Jvm運行，數據處理能力不如Mysql，Canal需要配置集群模式。一組Canal集群不能支持太多的事數據庫實例。
• 若是數據庫做了水平的分庫分表，原有Canal是不能識別為一類數據源，需要稍微修改部分代碼
• 建議Canal訂閱的Slave從機，因為Master是業務主庫，主庫承擔的業務系統職責太多
• Binlog日誌模式建議啟動Gtid，Canal訂閱的數據庫如果出現故障，需要基於此切換到其他數據庫。
• 數據輸出到Kafka，若數據庫是做了分庫分表的，需要修改Canal部分代碼。
• 設置Kafka分區鍵相同，保障相同數據變更順序。

Kafka中間件

使用Kafka作為中間緩存，主要基於以下幾個方面考慮
• 分區特性，Kafka支持分區，併發性能好，數據吞吐能力超過mysql，性能不會成為瓶頸
• 分區順序存儲，Kafka數據存儲是有順序的，設置好主鍵，保障Binlog變更順序
• 消費順序，客戶端消費Kafk數據，會基於Offset，按順序消費，保障Binlog變更順序
• 消費組，嚴格意義上講，Kafka並非熟悉消息隊列，應該算消息流，我們在上一篇文中討論數據模型映射需求，一個數據表可能會映射到多個索引，這就需要設置不同的消費組，保障多個消費組之間不衝突覆蓋，同樣的變更數據有多次重複消費

同步程序

同步程序當前基於Java自主開發，當前的主流同步工具不能很好支持自定義需求，主要包括兩大程序
同步任務調度程序

• 同步調度配置，配置同步任務，配置同步映射關係，DB到ES的映射，Kafka到ES的映射
• 同步調度分配，同步任務操作，啟動、停止、重新設置；同步任務分配，指定並行度等。

同步任務執行程序

• 執行任務，將數據從Kafka經過映射寫入到Elasticsearch中，主要由4大模塊組成
• Kafka模塊，拉取消費數據，記錄消費位置
• Mapper模塊，執行映射過程，數據表與索引映射，表字段與索引字段映射，生成指定的Json格式數據
• Elastic模塊，將Mapper生成好的Json數據提交到Elasticsearch中，成功則提交消費記錄位置，失敗則走另外邏輯
• Schedule模塊，基於線程級別維度執行同步任務，支持同步任務啟動暫停等操作，實時彙總同步任務的指標數據

數據同步全過程回顧

數據從DB更新到ES，中間經過多個環節，同步模式既有推送，也拉取，且多次結合完成。

• Mysql寫入到本地binlog，推送模式
• Canal讀取binlog寫入Kafka，先是拉取模式，後是推送模式
• Worker同步程序從Kafka讀寫數據，經過處理寫入到Elasticsearch，先是拉取模式，後是推送模式

注意事項

DB到ES實時同步整體項目鏈路很長，且涉及技術點較多，任意環境都會導致一些問題，有一些特別注意

• DB刷數據問題，由於DB是批量更新，後面幾個技術節點會出現部分性能瓶頸
• DB多表關聯深度問題，DB多表直接關聯最好的關係是1對1，主要ES映射也可以基於主鍵關聯更新，無需反向查詢
• ES高級類型限制問題，ES本身支持很多高級數據類型，但這些在同步程序中最好不要使用

遺留問題

項目推進過程中，遇到不少問題，有的已經解決，有的是無法解決，有的會改善解決

• 數據校驗，DB數據雖然同步到ES中，但目前是沒有有效的方法去校驗正確性的，傳統的方式校驗方式是隨機兩邊查詢對比，非常的低效，此處需要探討更好的數據比對方法，
• 數據修復，當數據發現不正確時需要自動能夠修復，但由於數據校驗的低效，數據修復的準確性也有待考量
• 技術演進，數據同步程序是自主基於Java開發，但做了很多非業務時間工作，程序大量的工作在調度且涉及很傳統，考慮引入Flink平臺，由平臺負責底層資源掉度，上層只需配置同步映射，當前正在測試中。

結語

經驗總結

DB到ES數據實時同步，前後經過的時間很長，經過了好幾次的技術方案演進，未來還有很大的優化改進空間。
最終進化到CDC這種方案也是基於已有的經驗分享（參考馬蜂窩技術博客），設計思路一樣，技術實現不一樣。
數據同步整體技術實現中間環節很多，任意技術點都需要了解透徹，否則會出現很多致命事故，需要多人團隊協作完成。

CDC並非新概念，幾乎所有數據庫產品都支持，如下：

• Postgresql 有 Logical Decoding
• Sqlserver 有 Change data capture 和 Change Tracking
• Oralce 有 Redo log和 Oralce Golden Gate
• Mongodb 有 Replicate sets
• Elasticsearch 有 Translog

若之後遇到類似數據實時同步需求，優先選擇CDC技術方案，我們需要更強大的數據實時交換平臺，歡迎討論一起幹。

聲明：本文由原文作者“李猛”授權轉載，對未經許可擅自使用者，保留追究其法律責任的權利。

【阿里雲Elastic Stack】100%兼容開源ES，獨有9大能力，提供免費X-pack服務（單節點價值$6000）