9年互聯(lián)網(wǎng)/大數(shù)據(jù)領(lǐng)域研發(fā)、架構(gòu)經(jīng)驗，2015年加入京東物流，主要負(fù)責(zé)大數(shù)據(jù)相關(guān)架構(gòu)與開發(fā)工作。

　　摘要： 數(shù)據(jù)蜂巢平臺是京東物流自主研發(fā)的分布式、高性能、高可用、支持異構(gòu)，離線和實時的大數(shù)據(jù)同步與管理平臺。關(guān)鍵技術(shù)：HA;離線與實時同步整合;binlog采集,存儲與訂閱;客戶端并發(fā)消費;一致性校驗與修復(fù);任務(wù)隔離。

　　目前已經(jīng)在京東物流系統(tǒng)中大規(guī)模應(yīng)用，比如單源和多源復(fù)制，從全國各地倉儲園區(qū)集群(上百個)實時復(fù)制到IDC，從mysql到ES，從mysql到cassandra等等。倉儲園區(qū)硬件、網(wǎng)絡(luò)環(huán)境復(fù)雜，數(shù)據(jù)蜂巢平臺需要考慮硬件設(shè)施和網(wǎng)絡(luò)故障的容錯性。議題主要分享平臺誕生的背景，使用的關(guān)鍵技術(shù)，架構(gòu)的演進過程，演進過程中所踩過的坑。

　　正文：

　　京東物流一線數(shù)據(jù)大部分都存在MySQL數(shù)據(jù)庫上，分布比較廣，包括國內(nèi)外園區(qū)庫房和IDC，這樣數(shù)據(jù)使用起來極不方便，各業(yè)務(wù)系統(tǒng)為了使用數(shù)據(jù)，開發(fā)出多種版本的同步工具，結(jié)果導(dǎo)致管理非常困難以及資源的浪費。這時就需要一個統(tǒng)一的平臺把這些數(shù)據(jù)管理起來。

　　架構(gòu)設(shè)計

　　關(guān)于數(shù)據(jù)同步，主要分為：批量同步、實時訂閱和實時同步。

　　批量同步： 采用sqoop的模式，把數(shù)據(jù)分片，然后進行多機并發(fā)復(fù)制，用以提升效率。

　　實時訂閱：使用消息隊列，即將binlog事件解析生成對應(yīng)的消息存儲在隊列中;

　　實時同步：通過客戶端去消費隊列將數(shù)據(jù)寫入目標(biāo)存儲，從而實現(xiàn)數(shù)據(jù)的實時同步。

　　如何中將以上三個功能整合到一個平臺?

　　首先，適配批量同步，參照sqoop, sqoop是借助于hadoop集群提交一個mr作業(yè)， 類似的，蜂巢系統(tǒng)也提供一個集群，將一個批量同步作業(yè)分為多task去執(zhí)行。實時訂閱也可使用這種思路，不同的是每一個task即對應(yīng)一臺mysql實例的binlog采集;同理實時同步也可分成多個task,每一個task即為一個消息隊列的消費客戶端。

　　架構(gòu)圖如下所示：

　　采用了經(jīng)典的master、slave結(jié)構(gòu)，每一個slave上可以跑三種對應(yīng)的任務(wù)。BatchWorker負(fù)責(zé)把MySQL數(shù)據(jù)批量同步到storage里面，這里的storage是一個抽象，不一定就是存儲之類的系統(tǒng)，也有可能是一些業(yè)務(wù)的處理。StreamWorker負(fù)責(zé)binlog的采集和消息隊列的維護，如果有訂閱需求，使用系統(tǒng)提供的客戶端Consumer就可以直接消費。

　　Pieworker則是Consumer的分裝，只不過這個客戶端由集群去管理維護。

　　BatchWorker結(jié)構(gòu)比較簡單，主要由Fetcher、Sinker和Storage(與整體架構(gòu)圖中的storage不同，此處為一個buffer)組成。Fetcher接口負(fù)責(zé)抽取數(shù)據(jù)，Storage負(fù)責(zé)緩存數(shù)據(jù)，Sinker接口負(fù)責(zé)數(shù)據(jù)的寫入。

　　StreamWorker模仿MySQL的主從復(fù)制機制， RelayLogTask負(fù)責(zé)binlog的抽取;HHLTask負(fù)責(zé)解析binlog，生成消息體，最后存在hhl中，hhl即為上文提到消息隊列。

　　客戶端: StreamJob除了采集binlog,維護消息隊列外，還提供了一個ClientServer模塊，用于接收客戶端的消費請求。當(dāng)Consumer需要消費的時候，請求會發(fā)送到ClientServer，ClientServer通過索引快速的定位到hhl文件的某一個位置，然后把數(shù)據(jù)讀取出來，然后經(jīng)過客戶端的指定過濾規(guī)則進行過濾，最終將消息體傳送給客戶端。

　　消息的位點由三部分組成的：Serverld對應(yīng)的就是MySQL的Serverld;binlogPosition是一個長整型數(shù)字，高32位為binlog文件下標(biāo)，低32位為文件內(nèi)部的位置;time為對應(yīng)的binlog事件產(chǎn)生的時間。

　　為解決性能問題，客戶端支持多種并行模式。

　　第一種是以事務(wù)為單位串行處理;

　　第二種是在第一種基礎(chǔ)上進行了一個簡單的優(yōu)化，比如：一個事務(wù)內(nèi)只對單表操作，那么這個事務(wù)是完全可以并發(fā)的，不同表間的事務(wù)順序并不會影響最終結(jié)果的一致性;但事務(wù)內(nèi)對多表操作就需要繼續(xù)串行。所以第二種并發(fā)模式就是不斷串行并行轉(zhuǎn)變的過程。

　　以上兩種消費模式是以事務(wù)為單位，以下兩種以行為單位

　　第三種是表級并發(fā)，將事務(wù)拆分為多個行級操作，同表操作由同一線程完成，保障同一表操作的有序性;

　　第四種是行級并發(fā)，主要是把MySQL的數(shù)據(jù)同步到NOSQL上時使用，關(guān)系型數(shù)據(jù)庫同步時局限性較大，因為行級并發(fā)只保障同一主鍵的操作有序，而關(guān)系型數(shù)據(jù)庫會存在多個唯一約束，這樣即使保障了主鍵的操作有序也可能引起數(shù)據(jù)不一致。

　　集群的特性：作為一個集群都需要要保證高可用、數(shù)據(jù)本地性和負(fù)載均衡。高可用這一塊主要分為三部分：

　　MySQL： 由DBA保證數(shù)據(jù)庫的高可用，但當(dāng)Mysql主從切換時，binlog的位點是不一致的，此時系統(tǒng)通過Serverld的檢測發(fā)現(xiàn)該變更，然后通過時間在新的mysql實列上定位正確的binlog位點。

　　master(Queen)：基于Zookeeper完成Active角色的選舉

　　Bee(Slave)：Bee宕機后由Master將其運行的任務(wù)遷移到其他的Slave上。

　　數(shù)據(jù)本地性：每一個Bee在啟動時都配置了機房，分組等信息，作業(yè)提交時可以指定自己期望的運行位置，與hadoop,spark類似。

　　負(fù)載均衡： 每一個Bee會將自己的負(fù)載信息通過心跳發(fā)給Queen,queen進行作業(yè)調(diào)度時，會在滿足數(shù)據(jù)本地性的前提下選擇壓力最小的機器去運行新任務(wù)。

　　演進

　　HHL文件丟失：上文提到過，如果Streamworker的運行主機宕機，Master會把它遷移到另外一臺機器上，但是Streamworker采集解析的binlog存在本地，遷移后會引起數(shù)據(jù)丟失，解決這個問題通用的方案是多副本，但大數(shù)據(jù)量下的多復(fù)本會造成磁盤空間的浪費，尤其是在庫房環(huán)境下。

　　并且這些數(shù)據(jù)有一個特點，就是發(fā)生遷移時，雖然解析過的數(shù)據(jù)丟失，但是原始binlog都會在機器上保存(dba會保留n天的binlog數(shù)據(jù))，最終可以通過數(shù)據(jù)補全來保證數(shù)據(jù)不丟失。

　　上圖為第一版streamworker的擴展，最右面的可以認(rèn)為是這一組的主線程，虛線位置發(fā)生任務(wù)遷移，切換到了新的主機上，此時虛線左邊的數(shù)據(jù)全部丟失。如果有客戶端需要消費丟失的數(shù)據(jù)，服務(wù)端則啟動一組新線程，然后進行catchup，catchup會把丟失的部分補齊并提供給客戶端消費。

　　元數(shù)據(jù)： binlog是不記錄字段名等元數(shù)據(jù)的，而客戶端消費時需要。最簡單的方式是收集到binlog之后，去源庫上查詢，但在binlog采集延遲期間如果有ddl操作，會導(dǎo)致元數(shù)據(jù)不準(zhǔn)確。為解決該問題系統(tǒng)實現(xiàn)了一個快照模塊。

　　在StreamJob初次啟動的時候，把對應(yīng)的MySQL里面所有表都做一份快照，在此后的運行期間監(jiān)控DDL操作，當(dāng)解析到DDL操作時會將原快照取出生成一個復(fù)本，并在這個復(fù)本上應(yīng)用這個ddl，生成新的快照 。這樣系統(tǒng)可以保證任何時刻binlog對應(yīng)的元數(shù)據(jù)都是正確的，方便用戶使用。

　　客戶端：服務(wù)端并不記錄客戶端的消費位點，消費的位置由客戶端自行存儲。由于客戶端采用的是并發(fā)消費模式，消息又是嚴(yán)格有序的，此時位點記錄就必須保障每一個記錄下的位點之前的所有消息都被正確處理了，此處引入了一個環(huán)形提交隊列(具體實現(xiàn)與disruptor類似)。

　　當(dāng)連續(xù)的多個消息被正確處理，并達(dá)到記錄位點的間隔，此時提交隊列會將一個位點寫入對應(yīng)的存儲介質(zhì)。比如1，2，3，4，5，8，10，14被處理完成，位點提交間隔為5，則5位置對應(yīng)的位點被記錄，當(dāng)6，7，9被處理完成后10再被記錄。

　　以下主要為易用性的改進：

　　SQL: 用戶通過SQL描述需要同步哪些字段，同步條件等信息，服務(wù)端通過解析sql執(zhí)行對應(yīng)的同步邏輯。

　　Union: 用于處理多表合一的場景，通過加入來源標(biāo)識字段來解決唯一約束。

　　Join: 在同步的過程中完成寬表的加工，內(nèi)部通過緩存，布隆過濾器等優(yōu)化方案來提升性能。

　　一致性較驗：

　　系統(tǒng)提供了兩種模型：

　　第一種是使用pt_table_checksum的思路，比如要較驗一張表的數(shù)據(jù)，先把這張表的數(shù)據(jù)分成多個片段，然后對這些片段進行crc計算，并將結(jié)果和計算的范圍存儲到同一個數(shù)據(jù)庫的表里去，此時會觸發(fā)一個binlog事件，當(dāng)消費者消費到這個事件后重現(xiàn)該操作，通過比對計算結(jié)果值來確定數(shù)據(jù)是否一致。

　　這種方式一是侵入性比較強，需要在原庫上建對應(yīng)的比對結(jié)果表，二是需要加鎖，三是對延遲的要求很高，當(dāng)延遲較大時，消費端拿不到對應(yīng)的比對事件，將無法確定數(shù)據(jù)一致性。

　　第二種數(shù)據(jù)校驗的模型是基于BatchJob實現(xiàn)。

　　Fetcher抽取源和目標(biāo)雙方的數(shù)據(jù)，排序后通過storage把數(shù)據(jù)傳遞給sinker，Sinker根據(jù)用戶自定義的比較接口對數(shù)據(jù)進行比較，最后將差異通過Collector進行收集。這種比對方式的缺點是不能保證時間序列，在比對的時候數(shù)據(jù)是變更的，比對出來的結(jié)果可能并不是真正的差異。

　　此時我們需要修復(fù)比對結(jié)果，首先在比對開始時把消費端的位點記錄下來，比對完成后，如果有差異則從比對開始前記錄的位點進行binlog重放，通過分析binlog中操作，對差異進行修復(fù)，輸出最終結(jié)果。

　　該比對模型最大的問題是需要抽取比對雙方的數(shù)據(jù)，對帶寬占用較大，為減少網(wǎng)絡(luò)傳輸，內(nèi)部會對數(shù)據(jù)進行初步的篩選，將要比對的數(shù)據(jù)分成多個片段，在存儲端對每個片段進行md5計算，fetcher只收集md5值 ，只有雙方md5值不同時才將真正的數(shù)據(jù)抽取到計算端進行比較。

　　修復(fù):

　　系統(tǒng)提供兩種修復(fù)方式：

　　一是基于binlog事件傳遞，當(dāng)比對出差異數(shù)據(jù)后，只需要在源庫上對差異數(shù)據(jù)進行一個偽操作(比如更改update_time字段)，觸發(fā)binlog事件的產(chǎn)生，消費端收到該事件即可修復(fù)錯誤數(shù)據(jù)(事件對應(yīng)的消息體內(nèi)包含了所有對應(yīng)字段的值，并不只限于update_time字段)，該方式缺點是受延遲的影響，同時還需要源庫的寫權(quán)限。

　　二是直接修復(fù)，即直接在目標(biāo)上將差異數(shù)據(jù)修正。在修復(fù)數(shù)據(jù)時需要加鎖或暫停同步，避免并發(fā)問題。

　　資源的隔離： 系統(tǒng)默認(rèn)只提供了常用的同步功能，當(dāng)默認(rèn)實現(xiàn)無法滿足客戶需求時，用戶需要自行編寫代碼來實現(xiàn)對應(yīng)的接口來完成他們的邏輯。Bee開始使用的線程模型，即每一個Bee可以看做一個線程池，多個任務(wù)都在同一線程池內(nèi)運行，而用戶自定義代碼又無法控制，導(dǎo)致不同任務(wù)相互影響，更有甚者任務(wù)結(jié)束后資源不釋放。

　　為保證資源隔離，將線程改為進程，將不同作業(yè)的任務(wù)交由不同的子進程去執(zhí)行，子進程啟動時會指定額定運行資源;所有子進程由Bee統(tǒng)一管理，Bee不再運行具體任務(wù)，作業(yè)結(jié)束后，Bee將未退出的子進程全部強制殺死。