背景

MySQL由于自身簡單、高效、可靠的特點，成為小米內部使用最廣泛的數據庫，但是當數據量達到千萬/億級別的時候，MySQL的相關操作會變的非常遲緩；如果這時還有實時BI展示的需求，對于mysql來說是一種災難。

為了解決sql查詢慢，查不了的業務痛點，我們探索出一套完整的實時同步，即席查詢的解決方案，本文主要從實時同步的角度介紹相關工作。

早期業務借助Sqoop將Mysql中的數據同步到Hive來進行數據分析，使用過程中也帶來了一些問題：

雖然Sqoop支持增量同步但還屬于粗粒度的離線同步，無法滿足實時性的需求
每次同步Sqoop以sql的方式向Mysql發出數據請求也在一定程度上對Mysql帶來一定的壓力
同時Hive對數據更新的支持也相對較弱

為了更有效地連接前端業務數據系統（MySQL）和后端統計分析系統（查詢分析引擎），我們需要一套實時同步MySQL數據的解決方案。

小米內部實踐

如何能夠做到數據的實時同步呢？我們想到了MySQL主從復制時使用的binlog日志，它記錄了所有的 DDL 和 DML 語句（除了數據查詢語句select、show等），以事件形式記錄，還包含語句所執行的消耗時間

下面來看一下MySQL主從復制的原理，主要有以下幾個步驟：

master（主庫）在每次準備提交事務完成數據更新前，將改變記錄到二進制日志(binary log)中
slave（從庫）發起連接，連接到master，請求獲取指定位置的binlog文件
master創建dump線程，推送binlog的slave
slave啟動一個I/O線程來讀取主庫上binary log中的事件，并記錄到slave自己的中繼日志(relay log)中
slave還會起動一個SQL線程，該線程從relay log中讀取事件并在備庫執行，完成數據同步
slave記錄自己的binlog

binlog記錄了Mysql數據的實時變化，是數據同步的基礎，服務需要做的就是遵守Mysql的協議，將自己偽裝成Mysql的slave來監聽業務從庫，完成數據實時同步。

結合小米內部系統特點，構建了Mysql數據同步服務–-LCSBinlog，作為一種獨立的數據接入方式整合在Talos Platform中，Talos Platform作為大數據集成的基礎解決方案，以自研消息隊列Talos為數據總線，連接各種系統為主要目標，提供豐富的數據Source輸入和數據Sink輸出，并且Talos天然支持流式計算，因此業務可以充分利用Talos Platform互聯互通的特性，并結合自身的業務需求實現更加高階的業務場景。

上圖是Talos Platform中的整體流程架構，其中標紅部分是目前LCSBinlog在小米內部使用最廣泛的一條鏈路：Mysql ---> Talos ---> Kudu ---> BI，數據同步到kudu后借助Sparksql查詢引擎為上層BI系統提供即席查詢服務，Kudu和Sparksql的整合細節可以參見往期內容：告別”紛紛擾擾”—小米OLAP服務架構演進

LCSBinlog服務的主體架構

服務一共有兩種角色

Master ：主要負責作業的調度，

Worker：主要完成具體的數據同步任務

在Worker上運行兩種作業：

BinlogSyncJob：每一個mysql庫都會對應這樣一個Job，將binlog日志完整地寫入到服務創建的Talos topic中
MysqlSyncJob：同步歷史數據，消費binlog數據，過濾特定庫表數據實時同步至用戶配置的topic中

服務整體依賴于Zookeeper來同步服務狀態，記錄作業調度信息和標記作業運行狀態；在kudu表中記錄作業同步進度

控制流程如下：

Worker節點通過在Zookeeper上注冊告知自己可以被調度
通過在Zookeeper上搶占EPHEMERAL臨時節點實現Master的HA
用戶在融合云（Web）上注冊BinlogSource同步任務
Master周期性從配置服務讀取Binlog同步作業配置
Master更新Zookeeper中的調度信息
Worker節點根據Zookeeper上的調度信息啟動新分配任務，停止配置失效任務；作業啟動后完成數據實時同步并周期性將同步進度記錄在kudu中
服務上報監控信息到Falcon平臺，作業異常退出發送報警郵件

如何保障數據正確性

>>>>順序性

用戶配置的每一個BinlogSource 都會綁定一個Talos的topic，在進行消費的時候需要保證同一條mysql記錄操作的順序性，消息隊列Talos是無法保證全局消息有序的，只能保證partition內部有序。

對于配置分庫分表或者多庫同步任務的BinlogSource，服務會根據庫表信息進行hash，將數據寫入相應的partiton，保證同一張表的數據在一個partition中，使得下游消費數據的順序性；

對于單表同步的作業目前使用一個partition保證其數據有序。

>>>>一致性

如何保證在作業異常退出后，作業重新啟動能夠完整地將mysql中的數據同步到下游系統，主要依賴于以下三點

服務會記錄作業同步的offset，重啟后從上次commit的offset繼續消費
Binlog數據的順序性保證了即便數據被重復消費（未commit的數據），也能對同一條記錄的操作以相同的順序執行
下游存儲系統kudu，Es ，redis基于主鍵的操作能夠保證binlog重復回放后數據的最終一致性

應用場景

有了這份數據我們可以做些什么事情呢，本節例舉了幾種常見的應用場景

>>>>實時更新緩存

業務查詢類服務往往會在mysql之上架設一個緩存，減少對底層數據庫的訪問；當mysql庫數據變化時，如果緩存還沒有過期那么就會拿到過期的數據，業務期望能夠實時更新緩存；

利用binlog服務，根據策略實時將數據同步到redis中，這樣就能夠保證了緩存中數據有效性，減少了對數據庫的調用，從而提高整體性能。

>>>>異步處理，系統解耦

隨著業務的發展，同一份數據可能有不同的分析用途，數據成功寫入到mysql的同時也需要被同步到其他系統；如果用同步的方式處理，一方面拉長了一次事務整個流程，另一方面系統間也會相互影響

數據在mysql中操作成功后才會記錄在binlog中，保證下游處理到時的一致性；使用binlog服務完成數據的下發，有助于系統的解耦

關于異步處理，系統解耦在消息隊列價值思考一文中有更深入的解讀

>>>>即席查詢的BI系統

就如文章開篇提到的，mysql在一定場景下的性能瓶頸，mysql數據同步到kudu后可以借助sparksql完成性能的提升

因為同樣是sql接口，對使用者的切換成本也是較低的，數據同步到更適合的存儲中進行查詢，也能夠避免因大查詢而對原mysql庫其他查詢的影響

目前小米內部穩定運行3000+的同步作業，使用binlog服務同步數據到kudu中；小米內部BI明星產品XDATA借助整套同步流程很好地支持了運營、sql分析同學日常統計分析的需求

如何使用Binlog數據

用戶接入數據的時候要求mysql庫開啟binlog日志格式必須為Row模式：記錄的是每一行記錄的每個字段變化前后的值，雖然會造成binlog數據量的增多，但是能夠確保每一條記錄準確性，避免數據同步不一致情況的出現

最終通過監聽binlog日志，LCSBinlog服務將數據轉換成如下的數據結構，寫入用戶注冊的Topic中，目前Sink服務使用SparkStreaming實時轉儲數據到kudu中，后續也將逐步遷移到Flink上以提升資源利用、降低延遲

業務用戶也可以根據我們提供的數據格式，實時消費Talos數據以實現更復雜的業務邏輯，下表為每一種數據操作，是否保存修改前后的列表

疑難雜癥

下面分享2個上線后遇到的有趣問題

>>>>數據不一致問題，業務使用唯一索引

業務接入一段時間后, 發現部分表會偶爾存在kudu表的數據條目數多于同步的mysql表的數據條目數，我們將多出來的數據與mysql產生的binlog日志經過一一對比，發現用戶在mysql表中設置了唯一索引，通過唯一索引修改了主鍵，而kudu中的數據是通過主鍵標識或更新一條記錄的，于是update操作變成了insert操作，這就造成了原來的1條記錄變成了2條。

解決辦法：對于這種類型的表，LCSBinlog服務會把一次Update操作轉換成一條Delete數據和一條Insert數據

>>>>Full Dump同步歷史數據時，客戶端超時

服務剛上線的時候，通過jdbc 執行sql的方式完成全量歷史數據的同步，在同步的過程中會發現dump任務會卡頓很長時間才會返回結果，當數據量很大會出現超時同步失敗的情況，會造成數據的延遲。調研后發現使用mysql官方jdbc在客戶端查詢數據的時候，默認為從服務器一次取出所有數據放在客戶端內存中，fetch size參數不起作用，當一條SQL返回數據量較大時可能會出現OOM

解決辦法：當statement設置以下屬性時，采用的是流數據接收方式，每次只從服務器接收部份數據，直到所有數據處理完畢。優化后歷史數據同步穩定運行，對mysql端的壓力也很小