MongoDB與Hadoop MapReduce的海量非結(jié)構(gòu)化數(shù)據(jù)處理方案

宋辰萱 孔祥文

（中國(guó)市政工程華北設(shè)計(jì)研究總院有限公司 天津市 300074）

1 MongoDB與MapReduce數(shù)據(jù)傳輸機(jī)制

MongoDB 與MapReduce 的整合主要由MongoDB Cluster、MongoDB-Connector for Hadoop 以及MapReduce Cluster 組成，MongoDB-Connector for Hadoop 充當(dāng)數(shù)據(jù)讀寫存儲(chǔ)的重要組成部分，MongoDB Cluster 承擔(dān)對(duì)非結(jié)構(gòu)化數(shù)據(jù)分片存儲(chǔ)的工作，MapReduce Cluster 負(fù)責(zé)并行計(jì)算的任務(wù)。[1]該連接器同時(shí)支持Pig以及Hive，從而可通過(guò)簡(jiǎn)單的腳本執(zhí)行相對(duì)復(fù)雜的MapReduce 工作流。

2 基于MongoDB分片技術(shù)的性能改進(jìn)

MapReduce 使用javascript 語(yǔ)法編寫，其內(nèi)部基于javascript V8引擎解析并執(zhí)行，javascript 語(yǔ)言的靈活性使mapreduce 可以處理更加復(fù)雜的業(yè)務(wù)場(chǎng)景。

MapReduce 主要分為以下幾個(gè)階段，如圖1所示。

Map：識(shí)別各種操作并將其分散至數(shù)據(jù)集群的各文檔中。

Shuffle：根據(jù)Key 值將數(shù)據(jù)集分組，同時(shí)為每一Key 值生成與其對(duì)應(yīng)的值表。

Reduce：數(shù)據(jù)表中的數(shù)據(jù)元素經(jīng)過(guò)反復(fù)處理解析，將數(shù)據(jù)表回寫至Shuffle，最終每個(gè)Key 有且只有一個(gè)數(shù)據(jù)表與其對(duì)應(yīng)，同時(shí)該數(shù)據(jù)表只存在一個(gè)元素。

Finalize：獲得數(shù)據(jù)最終計(jì)算結(jié)果并對(duì)該結(jié)果進(jìn)行加工整合。

圖2清楚地說(shuō)明 Map-Reduce 的執(zhí)行過(guò)程。

針對(duì)現(xiàn)有的分片技術(shù)提出性能改進(jìn)方案，如圖3所示。

2.1 設(shè)置合適的分片方式以及chunk size

由于數(shù)據(jù)執(zhí)行或更改時(shí)，頻繁拆分及遷移數(shù)據(jù)信息，輸入輸出資源會(huì)在splitting 和balancing 的過(guò)程中被大量消耗，因此數(shù)據(jù)遷移時(shí)，數(shù)據(jù)塊的大小設(shè)置對(duì)資源的開銷有一定的決定作用。數(shù)據(jù)塊大小默認(rèn)是64M，但實(shí)際運(yùn)用時(shí)為避免輸入輸出資源的大量消耗則靈活選擇分片方式，結(jié)合各自的業(yè)務(wù)特征，通常選擇基于范圍的分片方式或者基于Hash 索引的分片方式。

2.2 合理部署MongoDB分片集群Sharded Cluster

分散存儲(chǔ)Sharded Cluster 中的數(shù)據(jù)，實(shí)現(xiàn)系統(tǒng)性能最大化，當(dāng)用戶發(fā)起數(shù)據(jù)讀寫請(qǐng)求時(shí)，mongos 在Config Server 即時(shí)訪問(wèn)數(shù)據(jù)接口，獲取該數(shù)據(jù)集群中數(shù)據(jù)節(jié)點(diǎn)的路由信息，并將獲取的讀寫請(qǐng)求轉(zhuǎn)發(fā)至相關(guān)的數(shù)據(jù)分片中。用戶可通過(guò)其中一個(gè)或者多個(gè)mongo訪問(wèn)到整個(gè)數(shù)據(jù)集群，將對(duì)數(shù)據(jù)的讀取請(qǐng)求均勻分布于多個(gè)mongo中并將目標(biāo)數(shù)據(jù)分散存儲(chǔ)至分片中，從而達(dá)到負(fù)載均衡的目的[2]。

以下是連接分片集群的代碼（圖4）：

由此，為了是實(shí)現(xiàn)負(fù)載均衡，系統(tǒng)會(huì)自動(dòng)將用戶請(qǐng)求分散至數(shù)據(jù)集群中所有mongos 中。當(dāng)某個(gè)mongos 發(fā)生故障時(shí)，系統(tǒng)會(huì)自動(dòng)轉(zhuǎn)移故障，將應(yīng)用請(qǐng)求轉(zhuǎn)發(fā)至正常運(yùn)行的mongos 中。

2.3 平衡CAP

圖1

圖2

圖3

圖4

同一系統(tǒng)中一致性、可用性、分區(qū)容忍性往往不可同時(shí)兼得，這就要求我們?cè)诜植际郊軜?gòu)設(shè)計(jì)時(shí)須考慮取舍，犧牲其中一個(gè)或兩個(gè)要素來(lái)爭(zhēng)取另一要素的最大性能。在以上三種性能中，系統(tǒng)對(duì)分區(qū)容忍性及可用性依賴較高甚至不可或缺，而對(duì)一致性的需求相對(duì)較低。因此，對(duì)MongoDB 讀寫操作時(shí)需要因地制宜地進(jìn)行性能優(yōu)化配置，可犧牲一致性來(lái)爭(zhēng)取最大的可用性。為全面提高系統(tǒng)性能，對(duì)MongoDB 讀寫操作時(shí)設(shè)置用作數(shù)據(jù)寫入的數(shù)據(jù)節(jié)點(diǎn)，由此可將寫入的數(shù)據(jù)信息即時(shí)更新至其他用于備份的數(shù)據(jù)節(jié)點(diǎn)，而后的數(shù)據(jù)讀取可在其他的數(shù)據(jù)備份節(jié)點(diǎn)實(shí)現(xiàn)。

2.4 構(gòu)建內(nèi)存數(shù)據(jù)庫(kù)(Memory DB)與磁盤數(shù)據(jù)庫(kù)(Disk DB)的混合分區(qū)

為避免主機(jī)掉電或者系統(tǒng)重啟導(dǎo)致數(shù)據(jù)丟失，可設(shè)置讀取數(shù)據(jù)操作在Memory DB 中進(jìn)行，若在Memory DB 中找不到搜索的數(shù)據(jù)信息，再去訪問(wèn)Disk DB，執(zhí)行寫入操作時(shí)將數(shù)據(jù)或者內(nèi)存計(jì)算結(jié)果直接寫入Disk DB 則不會(huì)影響Memory DB 的訪問(wèn)速度，Disk DB與Memory DB 的數(shù)據(jù)定期同步，由此在確保數(shù)據(jù)完整性與準(zhǔn)確性的前提下突破了數(shù)據(jù)讀寫速度的瓶頸。傳統(tǒng)的分區(qū)模式使系統(tǒng)性能無(wú)法最大化地發(fā)揮，形成了水平擴(kuò)展行差，對(duì)數(shù)據(jù)存取速度及系統(tǒng)性能可謂是降維打擊，為打破這一技術(shù)僵局，所有的數(shù)據(jù)分區(qū)都將以Memory DB 和MySQL 關(guān)系數(shù)據(jù)庫(kù)的組織模式進(jìn)行分區(qū)，從而該混合分區(qū)的水平方向形成多個(gè)分區(qū)，垂直方向則形成二級(jí)數(shù)據(jù)庫(kù)分區(qū)，極大提高了數(shù)據(jù)訪問(wèn)的效率（如圖5所示）。

2.5 構(gòu)建有向無(wú)環(huán)圖以降低系統(tǒng)開銷

在MapReduce 框架中，存儲(chǔ)中間計(jì)算結(jié)果而后根據(jù)請(qǐng)求調(diào)入相關(guān)內(nèi)容，這種操作會(huì)數(shù)據(jù)大量冗余海量復(fù)制、磁盤輸入輸出以及序列化不必要的開銷?？蓞⒖紡椥苑植际綌?shù)據(jù)集（RDD）設(shè)計(jì)理念構(gòu)建有向無(wú)環(huán)圖（GDA），以管道化的方式將前一個(gè)操作的計(jì)算結(jié)果轉(zhuǎn)發(fā)至下一操作作為下一階段的計(jì)算輸入，省去中間計(jì)算結(jié)果的存儲(chǔ)，避免了海量數(shù)據(jù)復(fù)用及存儲(chǔ)，極大減少了序列化開銷。

2.6 計(jì)算本地化

數(shù)據(jù)節(jié)點(diǎn)放置計(jì)算代碼減少因數(shù)據(jù)移動(dòng)導(dǎo)致的資源消耗，同一數(shù)據(jù)范圍設(shè)置相匹配的Hadoop 及MongoDB，同一數(shù)據(jù)節(jié)點(diǎn)覆蓋相應(yīng)的數(shù)據(jù)節(jié)點(diǎn)與計(jì)算機(jī)節(jié)點(diǎn)且盡量chunk size 值相同，同集群同節(jié)點(diǎn)的部署方式便于實(shí)現(xiàn)計(jì)算向數(shù)據(jù)靠攏,數(shù)據(jù)節(jié)點(diǎn)直接計(jì)算極大程度上減少了數(shù)據(jù)中間計(jì)算存儲(chǔ)轉(zhuǎn)發(fā)導(dǎo)致的數(shù)據(jù)冗余。

3 基于MongoDB與Hadoop 的整合方案

如圖6所示，MongoDB 和Hadoop 的整合依賴于10gen 公司發(fā)布的中間件產(chǎn)品MongoDB Hadoop Connector，它負(fù)責(zé)將MongoDB和Hadoop 的整合依賴于10gen 公司發(fā)布的中間件產(chǎn)品MongoDB Hadoop Connector，使Hadoop 更全面地發(fā)揮其分布式運(yùn)算的能力。MongoDB Hadoop Connector 進(jìn)行資源整合，將MongoDB 取代了HDFS，在數(shù)據(jù)處理中扮演數(shù)據(jù)源的角色，數(shù)據(jù)集群在MongoDB中被切割為固定大小的數(shù)據(jù)塊，Mappers 根據(jù)數(shù)據(jù)的不同路由信息解析處理數(shù)據(jù)，經(jīng)過(guò)解析加工的數(shù)據(jù)結(jié)果通過(guò)Reducer 整合至期望的數(shù)據(jù)狀態(tài)將計(jì)算結(jié)果返回至數(shù)據(jù)源中。

在完整的數(shù)據(jù)整合流程中，Hadoop HDFS 沒(méi)有充當(dāng)任何角色，從而增加了Hadoop 與MongoDB 整合模式的可變性，在一定層面提高數(shù)據(jù)處理的效率。[3]

由圖5可見(jiàn)，Hadoop 與MongoDB 的整合框架由3 部分組成,MongoDB 與Hadoop 的整合可通過(guò)配置的方式提供以下四種方案：

（1）于HDFS 讀取目標(biāo)數(shù)據(jù)經(jīng)過(guò)數(shù)據(jù)處理解析，將解析結(jié)果回寫至HDFS。

（2）于HDFS 讀取目標(biāo)數(shù)據(jù)經(jīng)過(guò)數(shù)據(jù)處理解析，將解析結(jié)果回寫至MongoDB。

（3）于MongoDB 讀取目標(biāo)數(shù)據(jù)經(jīng)過(guò)數(shù)據(jù)處理解析，將解析結(jié)果回寫至HDFS。

（4）于MongoDB 讀取目標(biāo)數(shù)據(jù)經(jīng)過(guò)數(shù)據(jù)處理解析，將解析結(jié)果回寫至MongoDB。

4 總結(jié)

本文通過(guò)項(xiàng)目研究的實(shí)踐經(jīng)驗(yàn)，以分片與整合方式為研究視角，探索基于MongoDB 與Hadoop MapReduce 的海量非結(jié)構(gòu)化數(shù)據(jù)處理方案，以設(shè)置合適的分片方式以及chunk size、合理部署MongoDB 分片集群Sharded Cluster、平衡CAP、構(gòu)建內(nèi)存數(shù)據(jù)庫(kù)(Memory DB)與磁盤數(shù)據(jù)庫(kù)(Disk DB)的混合分區(qū)、構(gòu)建有向無(wú)環(huán)圖以降低系統(tǒng)開銷、計(jì)算本地化等方面進(jìn)行論述，并對(duì)這些改進(jìn)措施在實(shí)踐中的應(yīng)用情況進(jìn)行詳細(xì)分析。

圖5

圖6