作者介紹
李猛，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）
相關活動
更多折扣活動，請
通路阿裡雲 Elasticsearch 官網 阿裡雲 Elasticsearch 商業通用版，1核2G ,SSD 20G首月免費 阿裡雲 Logstash 2核4G首月免費