基于分布式系統的大數據隨機抽樣算法的實現

王磐++李勛++張濤

摘要：Hadoop是當前處理大數據環境的一套生態系統，按照層次結構為節點內的HDFS，根據該FS特性編寫的RPC，MapReduce框架，Yarn管理系統，其中各層次可細分或進行全層次結構的整合，如HBase關注于數據存儲方向，使用其中HDFS和RPC通訊對鍵值對數據進行轉換并實現分布式存儲，Spark關注于數據高速運算，通過高速緩存內存直接向上作用于RPC的機制和Yarn對資源的管理進行實時的分布式計算。該文根據在大數據中的快速進行有需求抽樣的需求，對存儲于HDFS中的大規模非結構化數據，RPC機制，及MapReduce中Map模塊做深入研究。

關鍵詞：Hadoop；大數據；隨機抽樣

中圖分類號：TP311 文獻標識碼：A 文章編號：1009-3044（2016）20-0009-03

1 概述

在對海量數據進行數據建模時，通常會遇到性能與耗時的問題，由于部分的數據建模算法是不可并行的，例如迭代，所以需要對數據進行預處理來大幅減少耗時。在傳統數據分析流程中，數據抽樣是一個較為通用的方法，但海量數據中不一定適用，因此本文提出了一種在Hadoop生態環境下的海量數據抽樣算法。本文將在第一章介紹技術背景，第二章闡述海量數據抽樣的可行性，第三章提供具體的技術路線及實現方法，最后進行驗證和總結。

1.1 背景

Hadoop目前已經更新在2.x的線路中，2.x引入最重要的機制是包含了MapReduce 2.0的Yarn架構.MRv2最基本的設計思想是將JobTracker的兩個主要功能，即資源管理和作業調度/監控分成兩個獨立的進程。在該解決方案中包含兩個組件：全局的ResourceManager（RM）和與每個應用相關的ApplicationMaster（AM）[1]。這里的“應用”指一個單獨的MapReduce作業或者DAG作業。[9]RM和與NodeManager（NM，每個節點一個）共同組成整個數據計算框架。RM是系統中將資源分配給各個應用的最終決策者。AM實際上是一個具體的框架庫，它的任務是“與RM協商獲取應用所需資源”和“與NM合作，以完成執行和監控Task的任務”?？傊哂辛己玫牟⑿心芰Φ痪邆鋵傮w的可見性。

1.2 抽樣在統計分析及數據挖掘中的意義

抽樣是非全面的對數據進行分析，從研究的總體中按隨機原則抽取部分單位作為樣本研究，并根據這部分數據的分析結果來推斷總體，以達到認識總體的一種統計方法。使用抽樣的條件：

1）用于不可能或不必要進行全面分析的總體特征的推斷。

2）用于分析模型的評價和驗證。

抽樣調查與大數據分析是有差別的，最大差別來源于數據表達上：前者抽樣調查，后者全面記錄；統計調查具有科學性、準確性、權威性。大數據具有不確定性、復雜性。就像望遠鏡讓我們能夠感受宇宙，顯微鏡觀測微生物一樣，成為新發明和新服務的源泉。大數據與統計數據相互佐證，差異較大。從大數據理論的最終要解決問題出發，大數據分析與傳統的統計分析完全不同，其分析內容不再是總體或群體特征，而是對個體特征的預測，是不需要進行抽樣的。大數據分析的目的是直接匹配答案，是不需要理解和解釋原因的分析理念。[2]現在環境存在兩個問題：1）能夠獲取的數據永遠僅是數據的一部分。2）數據的處理能力遠不及數據的產生能力。

所以在這個階段依然需要使用傳統的數學模型，進行抽樣，建模，擬合預測。同樣，存在這個問題還有大數據的排序。[3]在調查研究中發現，有許多利用大數據工具對數據進行排序的算法和相關研究。并且著名的TERASORT也依賴于Hadoop原生的隨機抽樣算法。

2 分布式系統下的抽樣可行性

2.1 非分布式數據的抽樣方法

通常需要抽樣的情況都知道總體的大小，某屬性的范圍，例如從A中根據A1屬性進行x隨機因子的分層隨機抽樣：1.計算size（A）；2.計算x*size（A），計算A1各值在x*size（A）中的占比；3.對每個標記A1值的子集進行抽樣，合并。

2.2 蓄水池抽樣算法

在大數據中，需要進行額外的一次數據遍歷才能知道總量，甚至在有實時增量的數據中，知道總量是不可實現的，通常的解決方法是蓄水池抽樣（Reservoir Sampling）：從N個元素中隨機的等概率的抽取k個元素，其中N無法確定。

算法流程描述為圖1：

2.3 MapReduce數據處理模型

在海量數據中不知道數據總體的大小，數據總體不在同一片存儲區，各存儲區容量可能不同，無法用傳統抽樣方法或蓄水池法。使用MapReduce處理數據包含以下幾個階段，以WordCount為例，圖2描述了在整個文章集中各個單詞出現數量的統計過程。

數據存儲在HDFS中是以BLOCK存在的，每個BLOCK有固定的大小及對應索引。在進入MapReduce程序處理前，形成建立KV值，根據當前集群的負載情況分成不定數的Split給Map處理。[6]Split記錄的只是該片在文件中的起始字節偏移量，數據的長度，以及該Split所包含的第一個BLOCK所在的主機地址，而對文件的操作是從“一個具體的Input”的層面來讀寫數據，而不是從“一個BLOCK”的層面來讀取數據。每臺主機在Map階段處理的僅是文件的某個片段，所以還需要一個對片段的結構描述，例如LineRecordReader，CSVRecordReader等，同樣也可自定義RecordReader。

單個Map-Reduce任務的執行過程以及數據輸入輸出的類型如下所示：

Mapper的四個方法是setup，map，cleanup和run。其中，setup和cleanup用 于管理Mapper生命周期中的資源，setup在完成Mapper構造，即將開始執行map動作前調用，cleanup則在所有的map動作完成后被調 用。方法map用于對一次輸入的key/value對進行map動作。run方法執行了上面描述的過程，它調用setup，讓后迭代所有的key /value對，進行map，最后調用cleanup。Combine，Reduce的周期基本一致，在簡單抽樣的過程中，僅需要利用其setup和cleanup階段進行數據掃描。

2.4 非等概率隨機命中方法

大數據分而治之的思想無處不在，隨機抽樣也是如此，大數據在存儲時已使用自身的算法對數據進行排序和索引。所以每個Split中的數據認為是均勻的，所以僅利用Map的抽樣方法描述為：

對于樣本數固定的抽樣，在setup階段利用近似估計法產生對應split中的隨機數據的概率表，在cleanup中寫入溢寫區。

樣本數為總體的百分比：

在setup階段利用使用泊松分布計算某樣本被選中的概率是否滿足成為候選者，并記錄下候選者。

setup階段完成可知道某個split對應的樣本總數。

cleanup階段計算需要補足或刪掉的候選者，由于候選數據是隨機抽取完成，再次加入極少量數據不影響大數據下的隨機性。[7]

3 抽樣算法的實現

context.getConfiguration 獲取抽樣比例λ，數據結構描述，抽樣方式參數組，偽代碼描述：

3.1 抽樣參數

一致性隨機抽樣可以使用偽隨機，在BLOCK存儲的數據是固定不變的使用隨機種子可保證多次抽樣結果一致。無放回的隨機抽樣中，需要保證每個獨立樣本如果可能被抽到，僅被抽到一次。上述算法描述即為無放回抽樣。有放回隨機抽樣在Map中利用KeyValue值，在setup中通過泊松分布生成第一組隨機數，該數字稱為飛鏢[8]，在cleanup中對所有的飛鏢再進行一次隨機“射擊”可保證每次“射擊”都是該split中的全體數據集。

3.2 驗證及性能優化

我們知道等概率的選取的樣本應該滿足正態分布，而正態分布的確定需要兩個參數 λ和均值，其中的均值是無法得知的，而正態分布是二項分布的連續，從個體樣本的選擇出發僅有兩種，備選和排除。在大數據中，假設數據量是足夠大的，因此抽樣過程可以描述為二項分布的極限情況，這正好符合泊松分布，參考圖3在λ越大時，其分布圖形趨近于正態分布，對應隨機抽樣分布：

根據德莫佛-拉普拉斯（De'Moivre-Laplace）中心極限定理，這列二項分布將趨近于正態分布。

合適的配置集群環境，數據處理速度影響最大的是非寄存器級的IO，所以分布的數據并非越多主機性能越優，在抽樣過程的處理中，考慮到CPU高速緩存的處理速度大概是磁盤IO的10～100倍，及系統正常的負載，使用每個節點同時處理的MAP數量控制在10個。使用LZO壓縮，MAP直接輸出并未使用中間輸出，不需要對緩存結果進行壓縮和網絡傳輸。調整Map和Reduce的數量到合適的值。根據分析邏輯使用Combiner使用最合適的Writable類型，無中間結果，重用Writable類型。

在節點內，各MAP再次分配對應64M的內存作為高速緩存，其中所有的local變量全部使用靜態[7]。分析Task的運行

4 結束語

本文對應工程僅使用平均200秒完成對10^8條數據，共200G的數據的遍歷，并僅使用一次遍歷完成對工程未知規模數據的抽樣。并通過100次20%的抽樣對抽樣的分布進行了擬合驗證，驗證結果表明完全符合泊松分布，測試曲線近似于正態分布，擬合度大于98%，符合在本文所處工程中的要求。

參考文獻：

[1] Tom White. 華東師范大學數據科學與工程學院， 譯. Hadoop權威指南3版[M]. 北京： 清華大學出版社， 2015.

[2] 李建江， 崔健， 王聃， 等. MapReduce并行編程模型研究綜述[J]. 電子學報， 2011（11）.

[3] 陳德華， 解維，李悅. 面向大規模圖數據的分布式并行聚類算法研究[C]//第29屆中國數據庫學術會議論文集（B輯）（NDBC2012）. 2012.

[4] ADAMS A， JACOBS D， DOLSON J，et al.The frankencamera： an experimental platform for computational photography. ACM SIGGRAPH 2010 papers，2010：1-12.

[5] DEAN J， GHEMAWAT S. Mapreduce： Simplified data processing on large clusters[J]. Communications of the ACM， 2008， 51（1）： 107-113.

[6] GHEMAWAT S， GOBIOFF H.The google file system[C]. ACM SIGOPS Operating， 2003.

[7] 李超越， 徐國勝. Hadoop公平調度算法的改進[C]//第十九屆全國青年通信學術年會論文集. 2014.

[8] 白永超， 付偉，辛陽. 基于Hadoop和Nutch的分布式搜索引擎研究與仿真[C]//第十九屆全國青年通信學術年會論文集. 2014.

[9] 范東來. Hadoop海量數據處理 技術詳解與項目實戰[M].北京：人民郵電出版社， 2015.