MongoDB與Hadoop MapReduce的海量非結構化數據處理方案

宋辰萱 孔祥文

（中國市政工程華北設計研究總院有限公司 天津市 300074）

1 MongoDB與MapReduce數據傳輸機制

MongoDB 與MapReduce 的整合主要由MongoDB Cluster、MongoDB-Connector for Hadoop 以及MapReduce Cluster 組成，MongoDB-Connector for Hadoop 充當數據讀寫存儲的重要組成部分，MongoDB Cluster 承擔對非結構化數據分片存儲的工作，MapReduce Cluster 負責并行計算的任務。[1]該連接器同時支持Pig以及Hive，從而可通過簡單的腳本執行相對復雜的MapReduce 工作流。

2 基于MongoDB分片技術的性能改進

MapReduce 使用javascript 語法編寫，其內部基于javascript V8引擎解析并執行，javascript 語言的靈活性使mapreduce 可以處理更加復雜的業務場景。

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

Map：識別各種操作并將其分散至數據集群的各文檔中。

Shuffle：根據Key 值將數據集分組，同時為每一Key 值生成與其對應的值表。

Reduce：數據表中的數據元素經過反復處理解析，將數據表回寫至Shuffle，最終每個Key 有且只有一個數據表與其對應，同時該數據表只存在一個元素。

Finalize：獲得數據最終計算結果并對該結果進行加工整合。

圖2清楚地說明 Map-Reduce 的執行過程。

針對現有的分片技術提出性能改進方案，如圖3所示。

2.1 設置合適的分片方式以及chunk size

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

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

分散存儲Sharded Cluster 中的數據，實現系統性能最大化，當用戶發起數據讀寫請求時，mongos 在Config Server 即時訪問數據接口，獲取該數據集群中數據節點的路由信息，并將獲取的讀寫請求轉發至相關的數據分片中。用戶可通過其中一個或者多個mongo訪問到整個數據集群，將對數據的讀取請求均勻分布于多個mongo中并將目標數據分散存儲至分片中，從而達到負載均衡的目的[2]。

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

由此，為了是實現負載均衡，系統會自動將用戶請求分散至數據集群中所有mongos 中。當某個mongos 發生故障時，系統會自動轉移故障，將應用請求轉發至正常運行的mongos 中。

2.3 平衡CAP

圖1

圖2

圖3

圖4

同一系統中一致性、可用性、分區容忍性往往不可同時兼得，這就要求我們在分布式架構設計時須考慮取舍，犧牲其中一個或兩個要素來爭取另一要素的最大性能。在以上三種性能中，系統對分區容忍性及可用性依賴較高甚至不可或缺，而對一致性的需求相對較低。因此，對MongoDB 讀寫操作時需要因地制宜地進行性能優化配置，可犧牲一致性來爭取最大的可用性。為全面提高系統性能，對MongoDB 讀寫操作時設置用作數據寫入的數據節點，由此可將寫入的數據信息即時更新至其他用于備份的數據節點，而后的數據讀取可在其他的數據備份節點實現。

2.4 構建內存數據庫(Memory DB)與磁盤數據庫(Disk DB)的混合分區

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

2.5 構建有向無環圖以降低系統開銷

在MapReduce 框架中，存儲中間計算結果而后根據請求調入相關內容，這種操作會數據大量冗余海量復制、磁盤輸入輸出以及序列化不必要的開銷。可參考彈性分布式數據集（RDD）設計理念構建有向無環圖（GDA），以管道化的方式將前一個操作的計算結果轉發至下一操作作為下一階段的計算輸入，省去中間計算結果的存儲，避免了海量數據復用及存儲，極大減少了序列化開銷。

2.6 計算本地化

數據節點放置計算代碼減少因數據移動導致的資源消耗，同一數據范圍設置相匹配的Hadoop 及MongoDB，同一數據節點覆蓋相應的數據節點與計算機節點且盡量chunk size 值相同，同集群同節點的部署方式便于實現計算向數據靠攏,數據節點直接計算極大程度上減少了數據中間計算存儲轉發導致的數據冗余。

3 基于MongoDB與Hadoop 的整合方案

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

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

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

（1）于HDFS 讀取目標數據經過數據處理解析，將解析結果回寫至HDFS。

（2）于HDFS 讀取目標數據經過數據處理解析，將解析結果回寫至MongoDB。

（3）于MongoDB 讀取目標數據經過數據處理解析，將解析結果回寫至HDFS。

（4）于MongoDB 讀取目標數據經過數據處理解析，將解析結果回寫至MongoDB。

4 總結

本文通過項目研究的實踐經驗，以分片與整合方式為研究視角，探索基于MongoDB 與Hadoop MapReduce 的海量非結構化數據處理方案，以設置合適的分片方式以及chunk size、合理部署MongoDB 分片集群Sharded Cluster、平衡CAP、構建內存數據庫(Memory DB)與磁盤數據庫(Disk DB)的混合分區、構建有向無環圖以降低系統開銷、計算本地化等方面進行論述，并對這些改進措施在實踐中的應用情況進行詳細分析。

圖5

圖6