基于MapReduce的并行CCBF大流識別算法

張惠民，馮林生

(陸軍裝甲兵學院， 北京 100072)

網絡行為監控是網絡安全管理的基礎性工作，將具有相同五元組的數據包進行歸類就得到了流(flow)數據，基于流(flow)數據的測量為網絡監控提供了新的途徑。流對象分布呈現出顯著的重尾分布特征，少部分的大流量對象形成了大部分的網絡流量[8,10]；在OC-768鏈路上，40 B大小的分組僅需8 ns的傳輸時間，集中式的處理方式難以滿足大規模網絡流量數據(以下簡稱“大流”)的處理需求。流對象之間在計算上不存在相關關系，非常適合于分布式并行計算。

MapReduce是Google提出的一種分布式的并行計算框架，Hadoop對MapReduce進行了開源實現，將復雜的計算任務抽象成多路的Map過程和Reduce過程[6]，便于并行化計算任務的實現。

本文在CCBF算法[1]的基礎上，設計了在Hadoop平臺運行的并行CCBF算法，并通過實驗驗證了算法的性能。

1 MapReduce模型的并行特性

在Hadoop框架中，MapReduce分布式處理框架由Map過程、Partition過程以及Reduce過程構成，如圖1所示。主函數負責完成對MapReduce作業正確運行所需參數配置，并顯式指定作業執行流程。Map函數定義為一個類(稱為Mapper類)。在Mapper類中可以定義setup、map和cleanup等方法。相應地，Reduce函數定義為一個稱為Reducer的類，在Reducer類中可以定義setup、reduce和cleanup等方法。map、reduce方法的一般形式表示如下：

map：( K1,V1)→list( K2,V2)

reduce：(K2,list( V2)) →list( K3,V3)

其中，(K1，V1)是數據節點存儲的數據分片中的記錄經過Hadoop框架解析后提供的鍵/值對表示，list( K2，V2)、list(K3，V3)分別是map和reduce方法輸出的鍵/值對列表。Combine函數和Map函數運行于TaskTracker節點，partition函數將map過程輸出的鍵值對定位到相應的reduce過程。

2 CCBF算法的并行化策略

2.1 CCBF算法思想

吳樺等[1]提出利用計數型布魯姆計數器[2,9]進行大流的識別記錄以及小流量對象的過濾，提出了CCBF算法,在數據量較大的情況下，該算法具有較小的錯誤率。

2.1.1 描述

CCBF算法采用計數型布魯姆過濾器對流(flow)進行計數，只有每個Hash表項的計數都超過閾值時，該流(flow)才被判斷為大流，并建立流的記錄，同時采用布魯姆過濾器進行大流存在的記錄，CCBF算法的結構如圖2所示。

2.1.2 算法

CCBF算法的具體描述如下：

while packet arrives

calculate H = h(1),h(2),…,h(K)

if H exist in Longflow_BF

{

Increase Longflow records

}

else

{

increase corresponding counters in Filter_CBF;

if min_counter>DemandValue{

decrease corresponding counters in Filter_CBF;

increase corresponding counters in Longflow_BF;

add new Long flow records;

}

}

end while

export flow record；

2.1.3 特性分析

令n為已經處理的報文的個數，k為Hash函數的個數，m為計數型布魯姆過濾器(CBF)結構中計數器的數目，j為大流的閾值，從而CCBF算法的誤判率為

(1)

CCBF大流識別算法以較低的時間和空間復雜度實現了高速鏈路大流的高精度識別，在數據量較大的情況下，該算法具有較小的平均錯誤率，在排除輸入、輸出影響的情況下，該算法可以應用于大規模主干網的流量分析。

2.2 基于MapReduce的并行化改造

2.2.1 描述

在基于流(flow)的網絡數據測量中，流對象之間在流的大小統計上不存在相關性；本文基于MapReduce的并行CCBF大流識別，在map過程中完成數據報基于流的特征提取，在partition過程中完成數據報基于流的分發，在reduce過程中完成長流的識別以及流長的計算。

2.2.2 map過程的設計

在MapReduce框架中，Map函數的輸入為p，p表示報文序列pi(i=1，2，…)。從報文序列中提取出五元組信息作為value，為實現良好的并行性，降低算法在各節點的耦合度，對五元組進行Hash操作，并將結果作為key，之后輸出〈key，value〉。

map函數的偽代碼如下所示：

input:p

output:〈key,value〉

key = hash(p.info)；

value=p.info；

2.2.3 partition過程的設計

在MapReduce分布式運行框架中，Partition過程用來完成對Map過程輸出的分區處理，Partition過程根據Reduce節點的個數完成對Map輸出的分區。因為在map過程中已經完成了hash操作，所以為提高算法的運行效率，在本文的算法中定制的Partition函數為

getpartition(K key,V value,int numReduceTasks)

{

return new long key;

}

2.2.4 reduce過程的設計

Reduce函數的輸入是〈key,[value]〉鍵值對。輸出為〈key′,value′〉鍵值對，其中key′為數據流的五元組標識符；value′數據流的長度，單位為數據包。Reduce函數首先通過Filter_CBF過濾掉小流，提取超過閾值的數據流存入鏈表，并用Longflow_BF結構標識長流是否存在，之后將鏈表中各節點存儲的五元組以及流長作為鍵值對輸出。

reduce函數的偽代碼如下所示：

input:〈key,[value]〉

output:〈key′,value′〉

while(value exist)

{

H = h1(value),h2(value),…hk(value);

if(H exist in Longflow_List)

{

update Longflow_List;

}

else

{

increase Filter_CBF;

if(Filter_CBFmin>threshold)

{

insert Longflow_BF;

insert Longflow_List;

decrease Filter_CBF;

}

}

}

node= Longflow_List->node;

key′=Longflow_Listi.info;

value′=Longflow_Listi.num;

context.write(key′,value′);

while(node->next != null){

node=node->next;

key′=node.info;

value′=node.num;

context.write(key′,value′);

}

3 算法分析

評價大流識別算法存在嚴格的標準，需要從算法運行的時間效率以及算法結果的正確性這兩個方面進行評價，在大流識別算法中這兩個方面是相互聯系互相制約的。其中算法的時間效率是主要方面，時間效率高的算法才可能被應用于大規模主干網上進行網絡流分析，時間效率將在后面的實驗中進行分析；算法的正確性決定了算法應用質量，從而決定了算法是否會被采用，下面從誤判率來分析算法的正確性。

算法的主體結構是使用計數型布魯姆過濾器進行大流識別判斷的，結合算法流程以及哈希沖突可知算法存在錯誤增加流長的誤判，使流長錯誤增加一次的誤判率為

(2)

這個概率是誤判一次的概率，根據計數型布魯姆過濾器的功能存在誤判兩次以上的概率，誤判兩次以上的概率遠遠低于誤判一次的概率，因而算法的誤判率取誤判一次的概率。分析誤判率公式，可以看出影響算法誤判率的4個參數：

N為布魯姆過濾器表示的元素個數，在Longflow_CBF中表示長流的總數，在Filter_CBF中表示網絡流的總數。隨著n的增大，算法的誤差增大，減小n的值就能相應地降低誤差，因此在固定的時間間隔內將布魯姆過濾器初始化，能夠降低誤判率。

K為Hash函數的個數。在n和m確定的情況下，為使算法的誤判率達到最小，k值為：

(3)

N表示在MapReduce運行框架中，reduce任務的個數。

M為布魯姆過濾器中表示集合元素的單元個數，m取值越大，誤判率越小，但是內存消耗也越大，因此需要在內存消耗和誤判率之間找一個平衡點。

將式(2)代入式(1)得：

(4)

從以上的分析可知算法的誤判率受硬件資源和被處理報文數量的影響。

4 實驗結果及分析

4.1 實驗環境與實驗數據

在實驗室搭建的Hadoop平臺上運行所有實驗。Hadoop實驗平臺由6臺機器組成，其中1臺為主節點，另外5臺為從節點。6臺機器的配置是：CPU的主頻頻率為8核2.40 GHz、32 GB內存和100 M網卡。操作系統為CentOS7。Hadoop版本是2.7.3，因為數據節點的采用的是8核CPU，所以reduce任務個數參數設置為8的倍數。6臺機器通過交換機和網線搭建一個局域網。

應用JAVA編程語言完成了文獻[1]中提出的CCBF算法以及本文提出的基于MapReduce的并行CCBF長流識別算法和不采用基于MapReduce的并行CCBF表流識別算法[1]。并進行了3組數據集實驗，分別測試了算法的運行時間、可擴展性以及加速比。實驗選取的數據源CAIDA-OC48數據集是在美國西海岸的一個OC48對等鏈路上采集的，采集于2012年9月，僅包括匿名的數據報頭部信息；CAIDA-OC192數據集是在CAIDA組織所監控的一個OC192鏈路上采集的，采集于2012年6月，僅包括匿名的數據報頭部信息，如表1所示，分別從中獲取2.9億個數據報文，使用從中獲得五元組(源地址、源端口、目的地址、目的端口、協議類型)[10]和二元組(源地址、目的地址)得到了4個數據集(CAIDA-OC48-5、CAIDA-OC48-2、CAIDA-OC192-5、CAIDA-OC192-2)。

表1 算法使用的數據源

4.2 采用與不采用基于MapReduce的并行CCBF大流識別算法性能對比

使用單機偽分布式方法搭建了一個Hadoop集群運行本文提出的算法，與運行CCBF算法的主機進行運行時間計算，結果如表2所示。

表2 算法執行時間 s

從表2可以看出，不采用基于MapReduce的并行CCBF大流識別算法的執行時間長于基于MapReduce的并行CCBF大流識別算法的執行時間，這是因為本文提出的算法是并行執行的，對多核CPU的利用率比串行執行的CCBF算法要高。

4.3 基于MapReduce的并行CCBF大流識別算法的加速比測試

文獻[4-6]給出了評價算法性能的重要指標加速比的公式，公式如下：

(5)

其中，t1、tn分別表示在1個節點和n個節點上的執行時間。

使用表1列出的OC48數據集驗證并行CCBF算法的加速比，取得的實驗結果如圖3所示。從圖3可以得出，算法具有較高的加速比。

4.4 基于MapReduce的并行CCBF大流識別算法的可擴展性測試

文獻[7]給出了評價算法性能的重要指標可擴展性的計算公式如下：

(6)

其中，w、p表示問題規模和集群規模，w′、p′表示擴展后的問題規模和集群規模。

為了便于測試算法的可擴展性，從表1選取OC192作為實驗數據集，實驗結果如圖4所示。實驗結果表明：并行CCBF算法能夠取得較好的可擴展性。

5 結論

CCBF算法是一種高效的大流識別算法，受單機計算性能的限制，CCBF算法對大數據環境下的大流識別存在執行時間過長、效率不高的問題。針對該問題，本文在MapReduce并行編程環境下提出了基于MapReduce的并行CCBF大流識別算法，并對算法進行了實驗分析，在多個數據集上的測試表明：基于MapReduce的并行CCBF大流識別算法具有較高的加速比和可擴展性。

：

[1] 吳樺,龔儉,楊望.一種基于雙重Counter Bloom Filter的長流識別算法[J].軟件學報,2010,21(5):1115-1126.

[2] BURTON H.Bloom.Space/time trade-offs in hash coding with allowable errors[J].Communications of the ACM,1970,13(7):422-426.

[3] FAN L,CAO P,ALMEIDA J,et al.Summary cache:A scalable Wide-area Web cache sharing Protocol[J].IEEE/ACM Trans.on Networking,2000,8(3):281-293.

[4] 周麗娟,王慧,王文伯,等.面向海量數據的并行KMeans算法[J].華中科技大學學報(自然科學版),2012,12(40):150-152.

[5] ZHAO Weizhong,Ma Huifang,He Qing.Parallel KMeans Clustering based on MapReduce[J].Lecture Notes in Computer Science.2009,5931:674-679.

[6] 陳興蜀,張帥,童浩,等.基于布爾矩陣和MapReduce的FP-Growth算法[J].華南理工大學學報(自然科學版),2014,42(1):135-141.

[7] 陳軍,莫則堯,李曉梅,等.大規模并行應用程序的可擴展性研究[J].計算機研究與發展,2000,37(11):1382-1388.

[8] 王風宇,郭山清,李亮雄,云曉春.一種高效率的大流提取方法[J].計算機研究與發展,2013,50(4):731-740.

[9] 田小梅,張大方,謝鯤,等.計數布魯姆過濾器代數運算[J].計算機學報,2012,35(12):2598-2617.

[10]朱應武,楊家海,張金祥.基于流量信息結構的異常檢測[J].軟件學報,2010,21(10):2573-2583.