一種正交分解大數據處理系統設計方法及實現

向小佳 趙曉芳 劉 洋 龔關俊 張 晗

1(中國科學院計算技術研究所 北京 100190)2 (北方工業大學計算機學院 北京 100144) (xiangxiaojia@ncic.ac.cn)

一種正交分解大數據處理系統設計方法及實現

向小佳1趙曉芳1劉 洋1龔關俊1張 晗2

1(中國科學院計算技術研究所 北京 100190)2(北方工業大學計算機學院 北京 100144) (xiangxiaojia@ncic.ac.cn)

MapReduce等計算框架的出現開啟了大數據處理新紀元，以Hadoop，Spark為代表的大數據處理系統具有大吞吐率、跨平臺、高可擴展的優勢，并得到廣泛應用.然而，為避免與具體的操作系統、硬件平臺綁定，這些系統的設計與優化集中在計算模型、調度算法等方面，無法充分利用底層平臺的優勢.提出了一種基于正交分解的大數據處理系統設計與優化方法，將系統分解為松耦合的多個功能正交的模塊，使存儲、處理功能分離出來，交給能夠利用底層平臺操作系統甚至硬件資源的存儲、執行引擎，原大數據系統退化為調度平臺；進而，提出基于鎖無關機制的存儲底層優化策略和基于指令超級優化的執行引擎底層優化策略.以此為指導，以Hadoop作為兼容和改進的對象，實現了原型大數據處理系統Arion.Arion既能保持Hadoop的跨平臺、高可擴展的優勢，又能消除任務執行的瓶頸，其本地化的設計與優化手段對非Hadoop平臺同樣有效.通過在原型系統上的實驗證明，Arion能夠提升大數據處理任務的執行效率，最高達7.7%.

大數據處理系統；計算框架；本地化；鎖無關；超級優化；執行引擎

網絡大數據的復雜性、不確定性、涌現性[1]給當前IT系統的架構、計算能力帶來了挑戰和機遇，催生了大數據處理框架.圍繞著這些計算框架，誕生了各種大數據處理系統.例如用于批量大數據處理的Google GFS與MapReduce[2],Nokia的Disco[3]、面向流式處理的Google Dremel[4],Microsoft的Dryad[5],Twitter的Storm[6],Yahoo的S4[7]等，學術界和開源社區也圍繞著面向批量大數據處理的Apache Hadoop、基于Hadoop的更具實時性的Impala*Cloudera Impala. https://github.com/cloudera/impala、伯克利AMP Lab的基于RDD[8]的、面向工作集迭代應用的Spark[9]展開了深入研究，國內的互聯網巨頭百度、阿里、騰訊等也在Hadoop等系統上部署了應用.

各類系統面向不同的應用，設計有針對性的計算模型、調度算法、數據結構，從而不斷演進.如Dremel，Storm等流式處理模型，相對于Hadoop，更適合海量流數據的即時查詢；而Spark則針對MapReduce模型不擅長的迭代處理和交互應用，提出了RDD內存數據集及相關迭代模型；Hadoop自身的計算框架由原本單一的MapReduce演化出了基于有向無環圖(directed acyclic graph, DAG)的更為靈活的Tez*Apache Tez. http://incubator.apache.org/projects/tez.html；Hadoop自身的調度系統也從單一的全局任務調度發展到了新一代的Yarn[10]，分離了JobTracker的資源管理與任務調度功能.

然而，由于大數據處理系統規模大，強調平臺無關性，避免與具體的操作系統、硬件平臺掛鉤，上述系統的演進都忽視了對底層平臺技術的利用.Intel中國研究院的NativeTask[11]通過設計外掛的計算引擎模塊，將部分Hadoop計算引擎內部的計算外延到Java虛擬機之外，取得了一定的本地化效果，思想值得借鑒，但還未充分發揮存儲結點、計算結點本地操作系統、硬件平臺的潛力.國內的百度公司也提出了Hadoop的C++擴展*Hadoop C++ Extension: A Framework for Making MapReduce More Stable and Faster. http://wenku.baidu.com/view/1859f988d0d233d4b14e69c8.html，通過使用類似Pipe的協議將Map和Reduce兩階段的Java執行邏輯替換為C++編寫并預編譯好的二進制可執行文件，向本地化邁進了一步，但其失去了中間邏輯表示的靈活性，同時本地化僅限于Map和Reduce的用戶邏輯，沒有考慮通用中間處理環節的本地化，也沒有深度挖掘代碼的優化空間.

我們首先提出一種基于正交分解的大數據處理系統設計與優化方法，主張大數據處理系統的設計應采用松耦合架構，能明確區分出功能正交的調度管理、數據存儲、任務執行三大功能模塊；與任務邏輯緊相關的數據存儲與任務執行應下沉到具體的硬件平臺去完成，大數據系統只負責最核心的資源與任務調度.進而，在此方法的指導下，我們提出了基于鎖無關機制的存儲底層優化策略，以及基于二進制指令超級優化的執行引擎底層優化策略.前者利用存儲結點操作系統的原語，構造鎖無關數據結構(lock free data structure)作為構造共享存儲的基石；后者基于大數據系統的應用針對性，采用超級優化(super optimization)方法，在執行邏輯中間表達或二進制代碼層面作離線靜態分析，利用編譯技術來發揮底層平臺潛力.

基于上述技術，以Hadoop作為兼容和改進的對象，我們實現了原型大數據處理系統Arion.Arion采用了松耦合的方式，將Hadoop的任務調度與存儲、計算引擎拆分開，通過通信協議連接；數據存儲下沉到ArionFS，這是一個采用C語言實現的大數據分布式文件系統，其中實現了基于鎖無關(lock free, LF)結構的元數據存儲，以及基于狀態機的文件數據鎖無關共享讀寫協議；任務執行則下沉到基于LLVM[12]的本地化執行引擎，能夠在代碼中間表示(IR)層面或二進制代碼層面采用類似窺孔(peep-hole)這樣的超級優化方法，提升執行引擎效率.

1 方法與設計

1.1 基于正交分解的大數據處理系統設計與優化方法

基于正交分解的設計優化方法原則有4點：

1) 大數據處理系統的設計應能明確區分出調度管理、數據存儲、任務執行三大正交功能模塊；

2) 各功能模塊松耦合，由類RPC通信協議互聯；

3) 大數據系統只負責最核心的資源與任務管理，工作于中間層，與平臺、語言無關；

4) 與任務邏輯緊相關的兩大功能模塊，數據存儲與任務執行，應外包到具體的硬件平臺去完成，且應平臺相關，以期充分發揮平臺潛力.

平臺相關是指利用物理節點的操作系統、硬件機制來設計與優化，與平臺底層軟硬件緊耦合.優點是能夠充分利用平臺資源，挖掘其潛力；缺點是某一平臺的優化成果，如優化二進制代碼，不能無縫遷移到其他平臺；但這并不影響整個系統的平臺無關性與可擴展性，這是由第3點原則保證的.

進而，我們提出了2條平臺相關優化設計策略：

1) 基于鎖無關機制的存儲底層優化策略.利用存儲結點操作系統的原語，構造鎖無關數據結構作為元數據共享存儲的基石；進而可將鎖無關機制擴展到文件數據的訪問流程中.

2) 基于指令超級優化的執行引擎底層優化策略.大數據系統有應用針對性，平臺執行引擎中的計算也具備一定的模式，有規律可尋，因此可采用超級優化的方法，在執行邏輯中間表示或二進制硬件指令層面作離線靜態分析，提取優化指令序列，創建優化數據庫，采用一次寫入多次讀取(write once read multi-time, WORM)模式，利用編譯技術來挖掘底層平臺的潛力.

1.2 Arion整體架構設計

Arion以Hadoop作為兼容和改進的對象，將數據存儲、任務執行與調度管理分離；以Hadoop pipe*Hadoop Pipes. http://www-01.ibm.com/support/knowledgecenter/SSGSMK_6.1.1/mapreduce_integration/map_reduce_hadoop_pipes.dita為基礎，使用C語言擴充了一套功能模塊間的通信協議；任務的調度管理工作于中間層，采用LLVM的中間表示語言IR作為任務執行邏輯的載體，與具體底層平臺、語言無關；數據存儲下沉到全C語言實現的分布式文件系統ArionFS，支持鎖無關元數據與數據訪問；任務執行則下沉到基于LLVM的本地化執行引擎，能夠在代碼中間表示(IR)層面或二進制代碼層面采用超級優化方法，提升執行引擎效率.

圖1為Arion的整體框架圖.一個高級語言(例如C++)書寫的MapReduce大數據處理程序的處理過程如下：1)該程序作為輸入，通過編譯前端(例如Clang*Clang: A C Language family frontend for LLVM. http://clang.llvm.org)編譯后轉變為中間表達IR文件；接下來，中間表達被提交給大數據處理框架，例如Hadoop，Hadoop會根據框架的定義生成大數據處理任務，交給JobTracker調度，具體工作派發給任務執行代理Task Tracker，而IR文件則通過Hadoop上傳到大數據存儲文件系統ArionFS中；2)任務執行代理通過類RPC的通訊協議ArionPipe與本地化執行引擎通信，后者會根據指令從ArionFS中獲取IR文件，從預編譯的函數庫中抽取本次處理會使用的連接件(見第2節)，并將它們組裝起來；3)LLVM本地執行引擎執行上述組裝好的任務，起始階段會通過連接件reader從ArionFS讀取需要處理的數據子集Split，結束階段通過writer向ArionFS回寫處理結果.

圖1中最上部虛線框中部分為退化Hadoop，主要負責調度管理，數據存儲與任務執行的功能被拆分出去；外部存儲通過加載ArionFS的函數庫接入，通信協議為ArionFS的基于狀態機的鎖無關共享讀寫協議(見2.3.2節)；外部任務的控制與狀態反饋則由Task Tracker任務執行代理接入，通信采用ArionPipe協議.

圖1的中部為基于LLVM的任務執行引擎及其適配層.執行引擎中，SuperOptimization指令級加速引擎依賴平臺技術和編譯超級優化技術，挖掘優化代碼片段，提升引擎效率；Pass Manager為LLVM的編譯階段管理結構；即時編譯(just in time, JIT)引擎為代碼執行的核心.適配層為針對框架(Hadoop)所編寫的便于大數據處理工作流無縫銜接的通用代碼庫，一般會被預編譯成JIT引擎可以執行的二進制代碼，或易于發布的IR形式，并于任務執行時按需動態加載.這些框架相關的大數據處理通用代碼在Arion中按功能封裝為類，后文稱之為連接件，典型連接件如用于Map和Reduce階段銜接的通用Partitioner.

Fig. 1 Arion architecture圖1 Arion框架圖

存儲如圖1的底層，即數據管理層.該層為全C語言實現的鎖無關大數據文件系統Arion FS.采用鎖無關機制可以有效解決數據訪問過程中的沖突，提高并發性.鎖無關機制包括基于OS原語的鎖無關元數據和基于狀態機的數據鎖無關讀寫協議兩部分.

數據管理層左部為ArionFS軟件模塊堆棧，底部為核心層，上部為接口層.核心層封裝了文件系統的名字空間布局管理(layout)、流管理、狀態機管理等功能，能夠直接調用OS本地系統調用，Libc函數庫；接口層高度模塊化，支持標準的Posix文件訪問協議及其上的Libc函數庫，同時提供二次開發用C/C++ API和Java API，其中Posix相關模塊在內核實現.

綜上，Arion采用面向平臺的本地化設計方法，對Hadoop做了正交化分解，形成了專職調度管理的退化版大數據框架、基于LLVM的任務執行引擎以及大數據存儲ArionFS的三大松耦合模塊架構.

1.3 數據存儲底層優化

分布式文件系統作為共享大數據的載體，其上的數據訪問具有高并發的特點.為了利用底層優化技術來應對海量高并發的負載，我們采用全C語言實現了一個原型大數據分布式文件系統ArionFS，其特點是：1)C語言實現，能夠直接利用操作系統的底層函數庫和資源.2)充分利用Lock Free結構.文件系統的元數據存儲采用Lock Free的Hash Table，能夠直接利用操作系統原語和數據結構設計來解決熱點元數據的并發訪問沖突，避免死鎖等問題；文件系統的海量數據訪問采用基于狀態機的鎖無關讀寫協議，解決并發訪問沖突，提高并發性.

ArionFS不支持類似HDFS的多副本流水線技術，其高可靠機制目前在卷、塊設備這一層級實現.

文件系統的元數據存儲采用了鎖無關Hash表(lock free Hash table, LFHT)，支持對元數據條目查詢、插入、刪除，這里的元數據主要是DENTRY條目(后文簡稱條目)，用來存儲文件名與具體數據分布等屬性的映射關系.相對于基于鎖的并發元數據訪問，實現鎖無關Hash表有3個特點：

1) 碰撞Hash值的條目置于同一Hash桶中，Hash桶組織為鏈表結構，基于CAS(比較與交換)原語來解決讀寫沖突；

2) 基于Pin來標識條目的所有權，支持批量刪除(Pin是一種所有權聲明工具，用來聲明對資源的占有，控制對資源的回收，以免出現一致性問題)；

3) 不同Hash桶采用動態分配隊列(dynamic allocated array)來組織，訪問文件條目速度為LogN，且空間動態分配刪除，節省存儲資源.

文件系統的數據訪問采用了基于狀態機的鎖無關讀寫協議，取消了共享文件并發訪問時的鎖操作(file lock).數據的鎖無關讀寫是基于CAS的鎖無關算法在狀態機層面的擴展.ArionFS中，客戶端與服務器端都有各自的運行時狀態機.每個狀態代表一個執行階段，伴隨著一段文件訪問的執行邏輯，邏輯執行的結果是事件，事件會觸發狀態機跳向下一狀態或者下一個子狀態機，循環往復.這里的狀態機可以類比于CPU，每個狀態所封裝的執行邏輯類比于指令，因此，對ArionFS的客戶端和服務器端而言，狀態所封裝的執行邏輯是原子的，這是在狀態機層面實現類CAS原語的前提.

鎖無關讀寫流程如圖2所示，圖2中通信的三方包括分布式文件系統的元數據服務器、客戶端、數據服務器.讀寫由客戶端發起，執行過程如左部的客戶端IO子狀態機所示：①客戶端首先進入初始化狀態；②根據要訪問的文件名或ID向元數據服務器發消息以獲得相應文件的分布等信息；③檢查文件屬性、訪問權限；④初始化代表訪問請求的消息；⑤通過專用信道向數據服務器發送代表訪問請求的消息，并等待回復；若回復為數據訪問成功，則接收返回數據或結果，并提交給下一狀態；否則根據一定策略，選擇重試或者放棄；⑥檢查返回結果；⑦如狀態機執行過程中發生故障，檢查IO錯誤，處理并退出子狀態機.

Fig. 2 Lock free IO state machine for data reading and writing圖2 文件數據鎖無關讀寫狀態機

數據服務器端的IO子狀態機如圖2右半部所示，其執行過程是：

1) 主狀態機執行監聽任務，等待各個客戶端的訪問請求，一旦接收到請求消息，轉入IO子狀態機；

2) 初始化狀態機，初始化相關數據結構，啟動流處理機制；

3) 分析請求類型，如果是讀請求，派送給“讀IO”狀態；如果是寫請求，派送給“拷貝并修改”狀態；

4) 讀IO狀態中，執行讀取操作，并將結果以消息的形式發送給客戶端；

5) 拷貝并修改狀態，數據會被拷貝出來1份，形成新版本，并按照寫請求來進行更新；

6) “結果檢查并提交或退出”狀態是原子的，對于讀請求結果檢查總是成功，對于寫請求，該狀態執行邏輯會檢查數據的版本，確認數據沒有被服務器的其它并發線程修改過，確認結果為真則提交修改并通知元數據服務器，確認結果為假則修改作廢并通知客戶端；

7) 如狀態機執行過程中發生故障，檢查IO錯誤，處理并退出子狀態機.

1.4 任務執行底層優化

大數據平臺具有應用針對性，平臺執行引擎中的計算也具備一定的模式，計算的規律性使得我們可以利用線下耗時的SuperOptimizaion編譯優化技術來構建具有應用針對性的優化執行引擎，從指令級別去加速計算.圖3所示為指令級加速引擎的框圖.

Fig. 3 Super optimization based execution engine framework圖3 Super Optimization指令級加速引擎框圖

圖3中部為執行引擎使用的輔助存儲：優化代碼數據庫和指紋Map.首先，引擎的優化以基本塊(basic block)為單位，即程序中可連續執行的指令序列的集合.代碼優化數據庫用來存放優化過后的基本塊；指紋Map則記錄了基本塊的Hash值與數據塊存放地址間的映射關系.

圖3下部為學習流程，功能模塊依序為離線采集模塊、超級優化模塊、后處理模塊.采集模塊以Basic Block，即程序中可連續執行的指令序列集合，為單位，結合部分啟發式的規則，來預先選取值得進一步優化的程序段.被選中的基本塊進一步被各個優化模塊處理.目前的原型實現中包括3種優化：常量合并(constant fold)、無用變量消除(unused variable suppression)、子表達式提取(sub expression extrac-tion).這些優化在其他工程中，如GCC，都是比較成熟的技術，我們這里優化的特點是：1)2階段，不僅在LLVM的中間表示(IR)階段作優化，而且在平臺選擇后，執行Target-Dependent的優化；2)窮舉模式的優化，對于無用變量消除、子表達式提取，對表達式采用黑盒法作輸入值的窮舉測試，通過觀測確定可以消除的變量或提煉表達式，雖然耗時，但由于在架構中屬于離線操作，所以不影響平臺的執行效率.這種優化方式不僅對平臺上用戶提交的大數據應用，甚至對于成熟的算法也具備優化的潛力.

學習流程后處理模塊以優化成功的基本塊為輸入，計算指紋，將指紋及優化成功后的基本塊中間表示或機器碼段分別存入指紋Map和優化代碼數據庫.

圖3上部為工作流程，功能模塊依次為在線輸入模塊、運行時采集模塊、指紋提取模塊、比較模塊、優化JIT引擎.采集模塊處理經在線輸入模塊導入的程序，提取其中的基本塊；指紋提取模塊以采集模塊提取的基本塊為輸入，計算其指紋；比較模塊將計算得到的指紋與指紋Map中的指紋作匹配，根據結果(命中與否)來控制開關，保證輸入JIT引擎的是最優的機器碼段.優化JIT引擎負責執行，對于中間表示代碼，JIT首先編譯基本塊為相應的機器碼段，進而將其交給CPU執行；對于機器碼段，則由CPU直接譯碼執行.

2 實 現

我們以Hadoop為基礎，采用我們的設計思路和優化方法進行改造，得到原型大數據處理系統Arion，其中，任務執行層位于大數據框架與硬件平臺的銜接處，是系統正交分解的關鍵.首先，執行層中的連接件是框架技術的外延，是封裝為類的大數據處理通用代碼庫，在大數據處理中會被執行引擎頻繁調用；其次，執行引擎向上承接用戶的大數據處理請求，向下驅動底層平臺，其選型與實現直接影響系統的整體性能和跨平臺特性.

圖1中部任務執行層的左邊為面向Hadoop MapReduce框架的連接件，通過選擇合適功能的連接件，才能無縫銜接Map和Reduce階段，形成大數據處理工作流.

連接件以類的形式被預編譯成JIT引擎可以執行的二進制代碼，或易于發布的IR形式，并于任務執行時按需動態加載.對于較常用的連接件，如Hadoop中的文件讀取類LineRecordReader(見Hadoop官方API)，需要預先編譯成所在平臺的二進制代碼，以動態鏈接庫的形式加載，這種方式更具性能優勢；對于不常使用的連接件，如TeraSort中的TotalOrderPartitioner(見Hadoop官方API)，會收集輸入數據的樣本來構建Trie樹，進而為Map的結果做分區；這是一種特殊邏輯，僅在TeraSort處理流程中使用，因此實現中該類以IR形式發布，便于增加靈活性，同時減小動態加載的大數據處理通用代碼庫的內存占用量.

連接件向上與調度管理層通過代理協議來通信和同步.實現中代理協議ArionPipe基于HadoopPipe來實現，擴充了HadoopPipe的命令，使其能夠支持更多平臺相關的連接件，如用于排序的連接件Sorter，能夠支持分布式緩存等等.ArionPipe目前是基于Socket的點對點協議，在拓撲和協議數據壓縮方面還有較大優化空間.

連接件通過調用庫函數的形式來集成ArionFS客戶端，進而能夠向下直接訪問ArionFS.如LineRecordReader,LineRecordWriter都分別集成了ArionFS客戶端，在大數據讀取和寫入時通過運行鎖無關IO狀態機來與ArionFS服務器通信.又如TotalOrderPartitioner通過集成ArionFS客戶端，能夠讀取緩沖的采樣文件，進而構建Trie樹完成分區任務；Kmeans大數據計算的Mapper用來完成浮點數的聚類，也能夠通過IO狀態機來讀取緩沖在ArionFS中的分類文件.

Arion的任務執行引擎基于LLVM實現，不但能夠利用平臺相關技術來優化大數據處理，還能夠保證系統的跨平臺特性.

3 實 驗

為了驗證Arion正交分解框架改造的正確性以及相對的性能提升，我們做了文件系統測試，性能擴展性測試，以及MapReduce任務的階段分析.

測試集包括類HiBench的4個大數據應用：

1) DFSIO.用于對分布式文件系統做讀寫性能測試，每個Map任務根據鍵值做對應文件的讀或寫操作，根據Value值確定文件大小，Reduce任務收集吞吐率等統計信息.

2) WordCount.用于提取大文件中出現的單詞并統計分析單詞出現的頻次.

3) TeraSort.該實驗采用TeraGen產生的1GB大文件(文件中每行記錄在百字節左右，總行數為千萬量級)，通過TotalOrderPartitioner做數據的采樣分區(采樣參數為100 KB)，進而排序.

4) Kmeans.該實驗實現了機器自動聚類，采用了循環迭代的算法，實驗中采用了10 KB的雙精度數據集，并設定了初始聚類參數.

實驗共采用了5臺同配置HP服務器、Intel Xeon 32核處理器、32 GB的內存、1 TB SATA硬盤；操作系統為CentOS 6.文件系統實驗、性能擴展性實驗中，1臺服務器固定用作Hadoop/Arion的JobTracker和任務提交、監測客戶端；另外1～4臺服務器用作任務執行TaskTracker.Case Study實驗中，共采用了3臺服務器，1臺用作JobTracker和客戶端，另外2臺用作TaskTracker.所有實驗中，分布式文件系統(HDFS/ArionFS)部署在參與當前實驗的服務器上.

3.1 文件系統測試

本實驗中，1臺服務器用作:1)MapReduce任務管理JobTracker；2)文件系統元數據節點Namenode；3)任務提交客戶端.其他1～4臺服務器用作：1)任務執行TaskTracker；2)文件系統數據節點DataNode.所有Map任務會發起對分布式文件系統的讀寫操作，形成對各個數據節點的IO壓力，Reduce任務則收集、分析各個Map任務的IO相關統計信息.

本實驗中，HDFS和ArionFS文件系統的基本分塊大小ChunkSize皆為64 MB；由于ArionFS暫不支持副本機制，將HDFS的副本數設置為1；實驗測試文件集設置為：10個文件，每個文件1 GB.每組實驗分別作了10次，根據平均數繪制實驗圖表.

由圖4(a)，對于寫測試，隨著分布式文件系統節點數的增加，無論是HDFS還是ArionFS，其執行時間有增加的趨勢，而吞吐率有減小的趨勢，這是由于DFSIO測試中，數據是由客戶端節點分發到各個數據節點的，隨著數據的分布范圍的增加，10GB實驗數據的分發帶來了網絡開銷，致使執行時間增加；另一方面，隨著節點數增加，并發IO也會帶來帶寬的聚合、性能的提升，由圖4(a)可知，4節點實驗的執行時間與3節點實驗相比，反而有所減少.

從完成時間上分析，1～4個節點的文件系統寫測試中，ArionFS的完成時間較HDFS分別減少了2.0%，2.7%，3.7%，3.1%；從吞吐率上分析，ArionFS的吞吐率較HDFS分別提高了2.2%，3.1%，1.0%，0.8%.綜上，ArionFS的平均完成時間較HDFS減少2.9%，ArionFS的平均吞吐率較HDFS提高1.8%.

Fig. 4 Distributed file system test based on DFSIO benchmark圖4 基于DFSIO測試集的分布式文件系統測試

由圖4(b)，對于讀測試，隨著分布式文件系統節點數的增加，無論是HDFS還是ArionFS，其執行時間變化不大，略有減少，這里執行時間記錄的是整個實驗從開始到結束的總時間，即包括文件讀階段，也包括Shuffle，Merge等銜接階段，以及最后的Reduce階段和統計信息分析階段；吞吐率方面，兩個文件系統均隨著節點數的增加而減小，這是由于數據分布帶來了網絡開銷，ArionFS在吞吐率上的表現要略優于HDFS.需要說明的是，本實驗中吞吐率與完成時間不成反比的原因是DFSIO實驗中吞吐率并非由完成時間計算得出，而是根據每個Map任務中的文件讀取時間來計算的.

從完成時間上分析，1～4個節點的文件系統讀測試中，ArionFS的完成時間較HDFS分別減少了2.0%，4.2%，1.8%，2.9%；從吞吐率上分析，ArionFS的吞吐率較HDFS分別提高了0.7%，7.7%，10.8%，4.7%.綜上，ArionFS的平均完成時間較HDFS減少2.7%，ArionFS的平均吞吐率較HDFS提高6.0%.

3.2 性能、擴展性測試

本實驗首先從性能上針對3種負載展開Hadoop與Arion的對比.實驗中，文件系統的基本分塊大小ChunkSize和大數據處理引擎的處理單元大小SplitSize皆為64 MB.

由圖5(a)，WordCount實驗中，單節點Arion在Vanilla Hadoop基礎上優化后，執行時間縮短3.0%；2，3，4個節點也分別縮短5.0%，3.0%，3.0%，平均縮短3.5%.根據圖5(b)(c)，TeraSort實驗中，相較Vanilla Hadoop，Arion的執行時間平均縮短7.7%；Kmeans實驗中，平均縮短3.5%.

Fig. 5 Performance and scalability test圖5 性能及擴展性測試

其次，由實驗結果可知，各個實驗中隨著節點數的增加，無論是Vanilla Hadoop還是Arion，其執行時間都是在縮短的，可見Arion在做正交分解和優化后，保留了原有大數據處理系統的良好擴展性.其中，Kmeans實驗后期，隨著節點的增加，性能改善較小，多節點時擴展性表現一般的原因是Kmeans隨著節點的增多，瓶頸主要集中在Reduce階段，包括從各個分節點調度、提取、排序數據，并重新計算分類中心點.

3.3 Case Study

本實驗中我們作典型MapReduce任務執行各階段的精細化分析.表1、表2中，依序(由下至上)記錄了如下9種大數據任務處理階段的執行時間：

1) 適配層的連接件Reader，如LineRecord-Reader，用于與ArionFS對接，通過運行鎖無關IO狀態機讀取分布式文件系統的數據；

2) Mapper，MapReduce框架的Map任務處理階段；

3) 連接件Partitioner，如3.1節介紹的Total-OrderPartitioner，用于對Mapper的輸出結果根據一定的算法做分區；

4) 連接件Combiner，用于對Mapper的輸出結果作部分合并；

5) 連接件Merger，對多個分布式的Mapper的輸出結果做歸并提取；

6) 連接件Sorter，在Reduce預處理階段，做多路Mapper提取數據的排序；

7) Reducer，MapReduce框架的Reduce任務處理階段；

8) 連接件Writter，如3.1節介紹的LineRecord-Writer，用于與ArionFS對接，通過運行鎖無關IO狀態機向分布式文件系統寫入數據；

9) Others，除去上述階段外的其他開銷，如Arion中LLVM引擎與上層Hadoop JVM虛擬機間的通信等.

本實驗皆采用1臺JobTracker服務器、2臺TaskTracker服務器的硬件配置，與Map相關的任務處理階段的執行時間為平均值.

表1為WordCount實驗，采用100 MB的文件，而計算引擎的Split Size為64 MB，共切分為2個Map任務.該實驗Arion的面向底層平臺的優化集中在Reader，Mapper，Reducer，Writer四個階段.如表1，Arion的Mapper階段經優化其執行時間較Hadoop減少7.9%，但是，在Others階段，Arion的通信、調度開銷較大，執行時間反而增加了5.0%.因此，總時間Arion僅比Hadoop減少了3.0%.

Table 1 Wordcount Stage Analysis表1 Wordcount執行階段分析

表2為TeraSort實驗，采用1 GB的文件，而計算引擎的Split Size為512 MB，共切分為2個Map任務.該實驗Arion的面向底層平臺的優化集中在Reader，Mapper，Partitioner，Merger，Reducer，Writer六個階段.如表2所示，相對于Hadoop，Arion的Mapper階段執行時間減少9.8%，Partitioner部分，即TotalOrderPartitioner，執行時間減少11.2%，Reducer部分減少14.4%，這幾部分提升較大與計算密集，引擎優化效率較高有關；但是，在Others階段，Arion的通信、調度開銷較大，執行時間反增29.5%.因此，總時間Arion僅比Hadoop減少了10.0%.

Table 2 Terasort Stage Analysis表2 Terasort執行階段分析

綜上，本實驗表明：Arion對大數據處理部分執行階段、讀寫階段的優化具有明顯效果，但也會引入額外的通信、調度開銷.

4 相關工作

大數據處理系統的發展是二維的.從橫向看，各個系統的框架主要圍繞著數據處理負載的特性來進化.Google的MapReduce[2]是典型的2階段處理框架，具備一定的通用性，但最初主要是針對搜索引擎網頁的批量抓取而設計.Nokia的Disco[3]系統也采用了針對批量大數據處理而設計的框架.Microsoft的Dryad[5]則是主要面向工作流處理負載的框架，這類框架還包括Google的Dremel[4]，Twitter的Storm[6]，Yahoo的S4[7]等.面向圖計算，Pregel[13]和開源的Giraph*http://giraph.apache.org是典型代表.針對MapReduce模型不擅長的迭代處理和交互應用，伯克利AMP Lab提出了基于RDD[8]內存數據集的迭代模型處理框架，并設計了內存大數據處理系統Spark[9].

從縱向階段上看，各大數據處理系統自身也是圍繞著框架、調度算法的升級而升級的.Hadoop自身的計算框架由原本單一的MapReduce演化出了基于DAG的更為靈活的Tez；自身的調度也從單一的全局任務調度發展到了新一代的Yarn[10]，分離了JobTracker的資源管理與任務調度功能.Mesos[14]是參照操作系統內核而設計的大數據資源管理系統，其自身也是圍繞著資源調度種類的增加、資源調度算法的升級而發展的，在Hadoop生態體系中，其作用類似于Yarn.

綜上，現有大數據處理系統的發展無論從那個維度看，計算模型的改進、框架和算法的升級是關注點，底層平臺相關技術的利用存在空白.Intel中國研究院的NativeTask[11]通過設計外掛的計算引擎模塊，將部分Hadoop計算引擎內部的計算外延到Java虛擬機之外；百度公司的Hadoop的C++擴展*Hadoop C++ Extension: A Framework for Making MapReduce More Stable and Faster. http://wenku.baidu.com/view/1859f988d0d233d4b14e69c8.html，通過使用類似Pipe的協議將Map和Reduce兩階段的Java執行邏輯替換為C++編寫并預編譯好的二進制可執行文件，都取得了一定的本地化效果，但都未充分發揮存儲結點、計算結點本地平臺的潛力.

大數據文件系統是大數據處理的基石，如Google公司的GFS[15]，Amazon的S3，Dynamo[16]，Apache的Cassandra*http://cassandra.apache.org，Hadoop中的HDFS[17]以及基于快速網絡的FDS[18]等，它們都提供類Key/Value的存儲抽象和基于副本的可靠性機制，為上層的大數據應用服務.新一代內存文件系統Tachyon[19]以內存數據集為主要抽象為上層提供大數據存儲服務，其中引入了Lineage[20]的抽象來實現內存數據的可靠性，大大減少了存儲空間占用，提高了寫入帶寬，但也引入了重計算的調度等問題.同樣采用Lineage抽象的還有BADFS[21]，該文件系統提供了顯示的描述式語言接口，使得內核能夠從用戶輸入獲得Lineage信息.由上述可知，大數據文件系統的主流聚焦于在本地系統上建立一層分布式抽象層，提供適合處理的數據模型和可靠性保障等服務，缺乏本地化相關領域的研究.

任務執行引擎需要運行時(managed runtime environment, MRE)支撐，這主要是由于運行時能夠為程序運行提供靈活性和安全性.成熟運行時，如SmallTalk[22]，Self[23]，JVM[24]都有嚴格的高級語言綁定，如type-safe的檢查，需要程序符合其MRE的設計.微軟的CLI[25]能夠支持多種語言，但對語言互操作性有嚴格限制，不支持非控制(unmanaged)語言的優化，且不開源.LLVM[12]是一個開源的編譯器設計框架，提供通用、基于靜態單一分配(static single assignment, SSA)模式的代碼中間表示，同時具備系列開源前端，支持多種高級語言，后端能做多遍優化，因此是本文選型方案.

5 結論和局限性及未來工作

我們提出一種基于正交分解的大數據處理系統設計與優化方法，主張將整個系統劃分為功能上互相正交的、松耦合的模塊，將數據存儲與任務執行下沉到具體的硬件平臺去完成，大數據系統只負責最核心的資源與任務調度.進而，在此方法的指導下，我們提出了基于鎖無關機制的存儲底層優化策略，以及基于指令超級優化的執行引擎底層優化策略.這種優化方式與主流的面向計算模型、框架、算法的發展趨勢并不矛盾，是正交、互補的關系，有效填補了技術空白.

基于上述技術，以Hadoop作為兼容和改進的對象，我們實現了原型大數據處理系統Arion，采用正交分解，將Hadoop的任務調度與存儲、計算引擎拆分開，通過通信協議連接；數據存儲下沉到C語言實現的鎖無關大數據分布式文件系統ArionFS中；任務執行下沉到基于LLVM的本地化執行引擎.經實驗證明，同等硬件配置下，Arion的性能優于Hadoop.

本技術局限在于:分解后下沉到平臺的數據存儲與任務執行引擎帶來的加速比依賴于負載模式、代碼特點及JIT引擎實現，對于數據分布均衡、處理邏輯簡單、優化代碼段執行頻率低的任務，如何提高收益，深挖平臺潛力是我們下一步工作的重點.另外，正交分解方法在Spark平臺上的應用也是未來工作之一，Spark平臺將調度交給Mesos、存儲交給HDFS等，已然符合正交分解思想，并為后續面向平臺優化做了鋪墊；其計算引擎層面的優化是難點，需要針對RDD的Transformation和Action執行邏輯中間表達作超級優化，充分考慮代碼特點及JIT引擎實現；存儲可以直接交給ArionFS.

[1]Wang Yuanzhuo, Jin Xiaolong, Cheng Xueqi. Network big data: Present and future[J]. Chinese Journal of Computers, 2013, 36(6): 1125-1138 (in Chinese)(王元卓, 靳小龍, 程學旗. 網絡大數據: 現狀與展望[J]. 計算機學報, 2013, 36(6): 1125-1138)

[2]Dean J, Ghemawat S. MapReduce: Simplified data processing on large clusters[J]. Communications of the ACM, 2008, 51(1): 107-113

[3]Mundkur P, Tuulos V, Flatow J. Disco: A computing platform for large-scale data analytics[C] //Proc of the 10th ACM SIGPLAN Workshop on Erlang. New York: ACM, 2011: 84-89

[4]Melnik S, Gubarev A, Long J J, et al. Dremel: Interactive analysis of Web-scale datasets[J]. Proc of the VLDB Endowment, 2010, 3(1/2): 330-339

[5]Isard M, Budiu M, Yu Y, et al. Dryad: Distributed data-parallel programs from sequential building blocks[C] //ACM SIGOPS Operating Systems Review. New York: ACM, 2007: 59-72

[6]Leibiusky J, Eisbruch G, Simonassi D. Getting Started with Storm[M]. Sebastopol: O’Reilly Media, Inc, 2012

[7]Neumeyer L, Robbins B, Nair A, et al. S4: Distributed stream computing platform[C] //Proc of the 2010 IEEE Int Conf on Data Mining. Piscataway, NJ: IEEE, 2010: 170-177

[8]Zaharia M, Chowdhury M, Das T, et al. Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing[C] //Proc of the 9th USENIX Conf on Networked Systems Design and Implementation. Berkeley, CA: USENIX Association, 2012: 2

[9]Zaharia M, Chowdhury M, Franklin M J, et al. Spark: Cluster computing with working sets[C] //Proc of the 2nd USENIX Conf on Hot Topics in Cloud Computing. Berkeley, CA: USENIX Association, 2010: 10

[10]Vavilapalli V K, Murthy A C, Douglas C, et al. Apache Hadoop yarn: Yet another resource negotiator[C] //Proc of the 4th Annual Symp on Cloud Computing. New York: ACM, 2013

[11]Yang D, Zhong X, Yan D, et al. NativeTask: A Hadoop compatible framework for high performance[C] //Proc of IEEE Int Conf on Big Data. Piscataway, NJ: IEEE, 2013: 94-101

[12]Lattner C, Adve V. LLVM: A compilation framework for lifelong program analysis & transformation[C] //Proc of Int Symp on Code Generation and Optimization. Piscataway, NJ: IEEE, 2004: 75-86

[13]Malewicz G, Austern M H, Bik A J C, et al. Pregel: A system for large-scale graph processing[C] //Proc of the 2010 ACM SIGMOD Int Conf on Management of Data. New York: ACM, 2010: 135-146

[14]Hindman B, Konwinski A, Zaharia M, et al. Mesos: A platform for fine-grained resource sharing in the data center[C] //Proc of Symp on Network System Design and Implementation. Berkeley, CA: USENIX Association: 2011: 22

[15]Ghemawat S, Gobioff H, Leung S T. The Google file system[C] //Proc of the 19th ACM Symp on Operating Systems Principles. New York: ACM, 2003: 29-43

[16]DeCandia G, Hastorun D, Jampani M, et al. Dynamo: Amazon’s highly available key-value store[C] //Proc of ACM SIGOPS Operating Systems Review. New York: ACM, 2007: 205-220

[17]Shvachko K, Kuang H, Radia S, et al. The Hadoop distributed file system[C] //Proc of the 26th IEEE Symp on Mass Storage Systems and Technologies. Piscataway, NJ: IEEE, 2010: 1-10

[18]Nightingale E B, Elson J, Fan J, et al. Flat datacenter storage[C] //Proc of the 10th USENIX Symp on Operating Systems Design and Implementation (OSDI 12). Berkeley, CA: USENIX Association. 2012: 1-15

[19]Li H, Ghodsi A, Zaharia M, et al. Tachyon: Reliable, memory speed storage for cluster computing frameworks[C] //Proc of the ACM Symp on Cloud Computing. New York: ACM, 2014: 1-15

[20]Bose R, Frew J. Lineage retrieval for scientific data processing: A survey[J]. ACM Computing Surveys, 2005, 37(1): 1-28

[21]Bent J, Thain D, Arpaci-Dusseau A C, et al. Explicit control in the batch-aware distributed file system[C] //Proc of Symp on Network System Design and Implementation. Berkeley, CA: USENIX Association, 2004: 365-378

[22]Deutsch L P, Schiffman A M. Efficient implementation of the smalltalk-80 system[C] //Proc of the 11th ACM SIGACT-SIGPLAN Symp on Principles of Programming Languages. New York: ACM, 1984: 297-302

[23]Ungar D, Smith R B. Self: The Power of Simplicity[M]. New York: ACM, 1987

[24]Lindholm T, Yellin F, Bracha G, et al. The Java Virtual Machine Specification[M]. Reading, MA: Addison-Wesley, 2014

[25]Miller J S, Ragsdale S. The Common Language Infrastructure Annotated Standard[M]. Reading, MA: Addison-Wesley, 2004

An Orthogonal Decomposition Based Design Method and Implementation for Big Data Processing System

Xiang Xiaojia1, Zhao Xiaofang1, Liu Yang1, Gong Guanjun1, and Zhang Han2

1(Institute of Computing Technology, Chinese Academy of Science, Beijing 100190)2(School of Computer Science, North China University of Technology, Beijing 100144)

Big data stimulates a revolution in data storage and processing field, resulting in the thriving of big data processing systems, such as Hadoop, Spark, etc, which build a brand new platform with platform independence, high throughput, and good scalability. On the other hand, substrate platform underpinning these systems are ignored because their designation and optimization mainly focus on the processing model and related frameworks & algorithms. We here present a new loose coupled, platform dependent big data processing system designation & optimization method which can exploit the power of underpinning platform, including OS and hardware, and get more benefit from these local infrastructures. Furthermore, based on local OS and hardware, two strategies, that is, lock-free based storage and super optimization based data processing execution engine, are proposed. Directed by the aforementioned methods and strategies, we present Arion, a modified version of vanilla Hadoop, which show us a new promising way for Hadoop optimization, meanwhile keeping its high scalability and upper layer platform independence. Our experiments prove that the prototype Arion can accelerate big data processing jobs up to 7.7%.

big data processing system; computing framework; localization; lock free; super optimization; excecution engine

Xiang Xiaojia, born in 1977. PhD, associate professor. Senior member of CCF. His main research interests include big data processing system, distributed storage, operating system, etc.

Zhao Xiaofang, born in 1965. PhD, professor, PhD supervisor. Her main research interests include computer architecture, data management, information security, etc.

Liu Yang, born in 1991. Master candidate. His main research interests include operating system, distributed system, information security, etc.

Gong Guanjun, born in 1993. Master candidate. His main research interests include operating system, network security, etc.

Zhang Han, born in 1993. Bachelor. Her main research interests include digital media, distributed storage, etc.

2015-12-09；

2016-07-19

國家自然科學基金項目(61202061,61202413)；中國科學院計算技術研究所創新課題項目(20146080) This work was supported by the National Natural Science Foundation of China (61202061, 61202413) and the Innovation Program of Institute of Computing Technology, Chinese Academy of Sciences (20146080).

TP391