由于僵尸网络威胁着互联网的安全，其检测方法也随着僵尸网络的发展而发展。流行的僵尸网络检测技术一般是通过网络流量分析实现的，如通过PageRank算法实现检测；通过网络通信图识别；利用神经网络算法识别僵尸网络。这些僵尸网络检测方法或者需要依赖外部系统提供信息，不能独立完成检测；或者由于网络流量信息量巨大，在单个服务器上完成检测的工作效率比较低^[1]。此外，僵尸网络具有流量大的特征^[2]，因此贝叶斯分类训练阶段需要对大量的网络数据集进行训练，用单一结点来进行检测僵尸网络将会遇到计算时间和计算资源瓶颈^[3]。为利用贝叶斯算法较高的准确性，基于云的Hadoop机制和MapReduce实现贝叶斯算法，该算法根据流量分析判断网络访问流量信息中是否存在僵尸行为^[4]。基于MapReduce检测僵尸网络的贝叶斯算法，把贝叶斯算法训练阶段的先验概率、条件概率和检测阶段的后验概率的计算并行化处理。该方案把捕获的网络流量利用云环境的贝叶斯并行算法进行分析处理，最终检测出僵尸网络^[5]。

1 检测僵尸网络的计算架构

图 1给出了云环境下检测僵尸网络的架构，由被测网络环境、云环境和代理服务器层三部分构成，这三部分协同完成僵尸网络的检测^[6]。每个被测网络中有若干台机器和一个核心交换机，连接一个代理服务器，代理服务器与核心交换机连接，主要负责网络流量的采集、解析、过滤并上传到云环境中。云的Hadoop收集并处理各个代理服务器上传的网络流量^[7]。

代理服务器运行tcpdump抓包工具来抓取网络数据包，且将抓取的数据包以十六进制格式的文件存储, 经过解析将文件变为可读的格式，以便程序分析处理。解析后将数据包存储在file中，不同协议的数据包格式不同，例如TCP协议的流量数据包格式为：序号|数据包到达时间|源IP地址|源端口号|目的IP地址|目的端口号|协议|数据包字节数|FIN|SYN|ACK|RST。

代理服务器将解析后的file上传到云的Hadoop中，以便用MapReduce并行化的贝叶斯算法进行分析处理。file中每一行表示一个数据包。不同被测网络的代理服务器将file上传到云环境的同一个指定文件夹，以便在云环境中对各个被测网络的流量集中分析处理，检测出所有被测网络中的僵尸主机^[8]。

为适合MapReduce计算模型处理，须将file中的数据包进行预处理。将抓取的网络流量信息数据包处理成以行形式存储的文件^[9]，每行信息形式为：序号|数据包到达时间|源IP地址|目的IP地址|TCP数据流|时间间隔平均值|时间间隔变化|数据包字节数|数据包个数平均值|持续时间平均值|类标签。其中，类标签值为0或者1来标明该条网络数据是否属于僵尸网络，本文设类标签值为0的网络访问为正常网络，否则为僵尸网络。

随机选择类标签值为0的正常网络信息行的2/3，再随机选择类标签值为1的僵尸网络信息行的2/3，这些行信息合成文件作为训练数据文件，剩余行作为检测数据文件。然后，将训练数据文件和检测数据文件分别按行分块，分块过程由Hadoop自动按64 MB大小作为一个数据块处理。

2 基于MapReduce的贝叶斯算法

贝叶斯算法进行MapReduce设计的基本思路是：贝叶斯训练阶段形成知识库，先验概率P(b)、P(n)和条件概率P(w_i|n)、P(w_i|n)、P(w_i|b)、P(w_i|b) (1≤i≤6)，其中b表示僵尸网络，n表示正常网络。其中w₁为TCP数据流, w₂为时间间隔平均值, w₃为时间间隔变化, w₄为数据包字节数, w₅为数据包个数平均值, w₆为持续时间平均值。计算正常网络和僵尸网络的先验概率对应一个MapReduce计算过程，即MapReduce1；对6个属性列既要判断是否为僵尸网络又要判断是否在阈值内，即每个属性有4个判断条件，因此需要求24个条件概率，计算这24个条件概率对应另一个MapReduce计算过程，即MapReduce2。贝叶斯检测阶段基于由26个概率构成的知识库，根据贝叶斯算法公式：

计算每条网络数据的后验概率^[10]，进行分类并判断是否为僵尸网络。检测阶段对应一个MapReduce计算过程，即MapReduce3。图 2描述了贝叶斯算法检测僵尸网络的MapReduce并行化实现方法。

2.1 MapReduce1的设计

Map1接收到的训练数据是被Hadoop处理形成的〈Key, Value〉对，形式为〈该行起始位置相对于文件起始位置的偏移量, 文本文件中的一行信息〉的信息。由于MapReduce1是计算贝叶斯的先验概率，只需用到Value的类标签属性，所以Map1将每行Value数据按空格分隔成字符串数组，取出数组最后一项，即类标签值。判断类标签值，若为0，输出中间结果〈Key1, Value1〉对的形式为〈“正常网络”，1〉；若为1，输出中间结果〈Key1, Value1〉对的形式为〈“僵尸网络”，1〉。并且，MapReduce框架每执行一次map()说明处理一行数据，通过累加统计训练数据总行数，以成员变量sum存储。Map1只是一个数据准备阶段，使Reduce1能在该准备数据上继续处理。Map1过程伪代码如下。

输入：Object、Text。

输出：Text、IntWritable。

map(Key、Value)

{StringTokenizer itr=new StringTokenizer(value.toString());

String[]temp=new String[9];

While(itr.hasMoreTokens())

  {temp[i]=itr.nextToken(); //属性字符串数组

     i++;

  }

Sum++; //网络数据总行数

If(temp[8].equals(0))//类标签0为正常网络

    Context.write(“正常网络”，1);

Else//表示为僵尸网络

    Context.write(“僵尸网络”，1);

}

经过Map1把分块的每行信息都处理成〈Key1，Value1〉形式的等待整体处理的中间文件输出，MapReduce框架将每个Map1输出的中间文件的结果〈“正常网络”，1〉或〈“僵尸网络”，1〉按照Key值(正常网络、僵尸网络)进行分组形成新的〈Key2, Value2〉对，形式为〈类标签值，{1，1, …, 1 }〉。

Reduce1接收到的信息为〈Key2, Value2〉。Reduce1的任务是对Key2相同的中间结果计数，若Key2值为“正常网络”，统计的Value2的行数为正常网络个数，并以成员变量sum_yes存储; 若Key2值为“僵尸网络”，统计的Value2的行数为僵尸网络个数，并以成员变量sum_no存储。并分别用sum_yes/sum、sum_no/sum计算得到先验概率P(n)和P(b)，并以成员变量sum_yes_p和sum_no_p存储。Reduce1过程伪代码如下。

输入：Text、IntWritable。

输出：Text、FloatWritable。

reduce(Key，Value)

{for(IntWritable val:Value)

  Sum1+=val.get(); //若Key为“正常网络”，统计的是正常网络数据行数；否则为僵尸网络数据行数

If(Key.equals(“正常网络”))

{sum_yes=sum1;//存储正常网络数据行数

  sum_yes_p=sum_yes/sum; //正常网络先验概率

}

Else

{sum_no=sum1;//存储僵尸网络数据行数

  sum_no_p=sum_no/sum; //僵尸网络先验概率

}

  Context.write(key, (float)(sum1/sum)); //输出先验概率

}

经过MapReduce1的处理，形成2个以成员变量sum_yes_p、sum_no_p存储的概率, 即正常网络先验概率和僵尸网络先验概率，构成知识库的一部分，供检测阶段使用。

2.2 MapReduce2的设计

Map2接收到的信息与Map1相同，是训练数据被Hadoop处理形成的〈Key, Value〉对, 形式为〈该行起始位置相对于文件起始位置的偏移量, 文本文件中的一行信息〉的信息。MapReduce2计算贝叶斯的条件概率，需用到Value的6个属性列及类标签值。因此Map2将每行Value数据按空格分割成字符串数组，取出数组的第3~9项w₁, w₂, …, w₆, 以及类标签值。首先判断类标签值是否为“0”，然后判断各属性是否在各自阈值内。若标签值为“0”且属性值在阈值内，输出中间结果〈Key3, Value3〉对的形式为〈“w_i|n”，1〉；若标签值为“0”且属性值在阈值外，输出中间结果〈Key3, Value3〉对的形式为〈“w_i|n”，1〉；若标签值为“1”且属性值在阈值内，输出中间结果〈Key3, Value3〉对的形式为〈“w_i|b”，1〉；若标签值为“1”且属性值在阈值外，输出中间结果〈Key3, Value3〉对的形式为〈“w_i|b”，1〉。Map2过程伪代码如下。

输入：Object、Text。

输出：Text、IntWritable。

map(Key，Value)

{StringTokenizer itr=new StringTokenizer(value.toString());

String[] temp=new String[9];

While(itr.hasMoreTokens())

{temp[i]=itr.nextToken(); //属性字符串数组

      i++;

}

If(Float.parsefloat(temp[2])>140 && Float.parsefloat(temp[2]) < 150)

    If(temp[8].equals(“0”))

      Context.write(“w₁|n”, 1);

    Else

      Context.write(“w₁|b”, 1);

Else

    If(temp[8].equals(“0”))

    Context.write(“w₁|n”, 1);

    Else

      Context.write(“w₁|b”, 1);

    //其他5个属性列训练过程同上。

}

经过Map2把分块的每行信息都处理成以〈Key3, Value3〉形式的等待整体处理的中间文件输出，MapReduce框架将每个Map2输出的中间文件的结果按照Key值(w_i|n、w_i|n、w_i|b、w_i|b)进行分组形成新的〈Key4, Value4〉对，形式为〈条件字符串，{1，1, …, 1 }〉。

Reduce2接收到的信息为〈Key4, Value4〉。Reduce2的任务是对Key4相同的中间结果计数，若Key4值为“w_i|n”，统计的Value4的行数sum1为w_i在阈值内且属于正常网络个数，利用sum1/sum_yes求得条件概率P(w_i|n), 并以成员变量wi_in_nomal存储；若Key4值为“w_i|n”，统计的Value4的行数sum1为w_i在阈值外且属于正常网络个数，利用sum1/sum_yes求得条件概率P(w_i|n), 并以成员变量wi_out_nomal存储；若Key4值为“w_i|b”，统计的Value4的行数sum1为w_i在阈值内且属于僵尸网络个数，利用sum1/sum_no求得条件概率P(w_i|b), 并以成员变量wi_in_unnomal存储；若Key4值为“w_i|僵尸”，统计的Value4的行数sum1为w_i在阈值外且属于僵尸网络个数，利用sum1/sum_no求得条件概率P(w_i|b), 并以成员变量wi_out_unnomal存储。Reduce2过程伪代码如下。

输入：Text、IntWritable。

输出：Text、FloatWritable。

Reduce(Key、Value)

{for(IntWritable val:value)

    {

      Sum1+=val.get(); //key为不同的条件字符串，则统计的是满足不同条件的网络数据行数

    }

//计算Key字符串长度length

//截取条件字符串Key最后2个字符msg_temp

If(msg_temp.equals(“n”))//若属于正常网络

{Float result=sum1/sum_yes; //计算条件概率

    If(Key.equals(“w₁|n”))

      w1_in_nomal=result; //成员变量存储属性列1在阈值内且属于正常网络条件概率

      Else if(Key.equals(“w_i|n”))

w1_out_normal=result; //成员变量存储属性列1在阈值外且属于正常网络条件概率

//其他5个属性列计算在阈值内、外属于正常网络的条件概率同上

}

Else

{Float result=sum1/sum_no; //计算条件概率

    If(Key.equals(“w₁|b”))

      w1_in_unnormal=result; //成员变量存储属性列1在阈值内且属于僵尸网络条件概率

    Else if(Key.equals(“w_i|b”))

      w1_out_unnormal=result; //成员变量存储属性列1在阈值内且属于僵尸网络条件概率

//其他5个属性列计算在阈值内、外属于僵尸网络的条件概率同上

}

      Context.write(Key, result); //输出条件概率

}

由于MapReduce2要对训练数据的6个属性列进行训练，每个属性既要判断是否为僵尸网络又要判断是否在阈值内，因此每个属性有4个判断条件。因此，经过MapReduce2的处理，形成24个条件概率分别存储在24个成员变量里，这与MapReduce1形成的2个成员变量存储的先验概率共同构成完整的知识库，可用于检测僵尸网络。

2.3 MapReduce3设计

Map3接收到的检测数据是被Hadoop处理形成的〈Key, Value〉对, 形式为〈该行起始位置相对于文件起始位置的偏移量, 文本文件中的一行信息〉的信息。MapReduce3要对6列属性全部检测，需用到Value的6个属性列。所以Map3将每行Value数据按空格分割成字符串数组，取出数组的第3~8项，分别为TCP数据流、时间间隔平均值、时间间隔变化、数据包字节数、数据包个数平均值、持续时间平均值。判断6个属性列的值是否在各自阈值内，若在阈值内，分别利用存储条件概率的成员变量wi_in_nomal、wi_in_unnormal计算后验概率；若在阈值外，分别利用存储条件概率的成员变量wi_out_nomal、wi_out_unnormal计算后验概率。并将每行网络数据的正常网络后验概率P(n|d)和僵尸网络后验概率P(b|d)一起输出。输出结果〈Key5, Value5〉对的形式为〈数据所在行数，P(n|d) P(b|d)〉。Map3过程伪代码如下。

输入：Object、Text。

输出：Text、Text。

map(Key，Value)

{StringTokenizer itr=new StringTokenizer(value.toString());

String[] temp=new String[9];

While(itr.hasMoreTokens())

    {temp[i]=itr.nextToken(); //属性字符串数组

      i++;

    }

    P₁=sum_yes_p; P₂=sum_no_p;

If(Float.parsefloat(temp[2])>140 && Float.parsefloat(temp[2]) < 150)

    {

      P₁=P₁*w1_in_nomal;

      P₂=P₂*w1_in_unnormal;

    }

Else if(Float.parsefloat(temp[2]) < 140‖Float.parsefloat(temp[2])>150)

       {

       P₁=P₁*w1_out_normal;

       P₂=P₂*w1_out_unnormal;

       }

//其他5个属性列检测过程同上。

       Line++; //统计所在行数

       Context.write(Line, P₁ P₂);

}

经过Map3把分块的每行信息都处理成以〈Key5，Value5〉形式的等待整体处理的中间文件输出，MapReduce框架将每个Map3输出的中间文件结果按照Key值(数据所在行数)进行分组后发送给Reduce3。

Reduce3接收到的信息为〈Key5, Value5〉。Reduce5的任务是逐行比较网络数据的P(b|d)和P(n|d)的大小。若P(n|d)>P(b|d), 判断该行网络数据为正常网络数据；否则为僵尸网络数据。Reduce3伪代码如下所示。

输入：Text、Text。

输出：Text、Text。

reduce(Key，Value)

{StringTokenizer itr=new StringTokenizer(value.toString());

       String[] temp=new String[2];

       While(itr.hasMoreTokens())

       {temp[i]=itr.nextToken(); //正常网络后验概率与僵尸网络后验概率

           i++;

       }

If(Float.parseFloat(temp[0])>Float.parseFloat(temp[1]))//比较

       Context.write(Key, “正常网络”); //判断

Else

       Context.write(Key, “僵尸网络”); //判断

}

3 实验结果与分析

本文实验中的被测网络环境为某校园网中一个子网的流量，该子网内主机约200台，白天的网络流量为150~200 Mbps。实验采集了某天数据，为测试本文提出并行化的算法性能，分别使用了2个不同时间段的数据集D₁和D₂。D₁解析后的文本文件1.6 GB，TCP数据包个数23 631 638。D₂解析后的文本文件0.8 GB，TCP数据包个数11 570 835。

实验一通过选取检测不同的特征向量个数，分析贝叶斯分类器的正确率。分类器由训练D₁数据集获得, 检测率通过分类器对D₂测试获取。具体实验结果如图 3所示。

根据图 3所示的实验结果，可以得到基于MapReduce的贝叶斯分类对于检测僵尸网络的正确率很高的结论，因此该方法是有效的。另外选取不同的属性列个数直接影响基于MapReduce的贝叶斯分类器的正确率，在一定程度上，检测属性列个数越多正确率越高。

实验二改变TCP数据流个数属性列的阈值，经过多次反复测试得到“正确率”分布如图 4所示。

从图 4中可以看到，如果把TCP数据流个数选为需要检测的特征向量，那么它的阈值可以选择为140~145。另外，如果选时间间隔值作为特征向量，同样可以测出它的阈值为595~605；那么，可以一一通过这样的方法获得每个属性的阈值以及它取到阈值时的最高准确率。

实验三通过改变训练、测试数据集大小，比较数据集大小分别对普通串行贝叶斯分类检测僵尸网络和MapReduce并行化的贝叶斯分类检测僵尸网络效率的影响。具体实验结果如图 5所示。实验A为普通串行贝叶斯检测僵尸网络、实验B为MapReduce并行化贝叶斯检测僵尸网络。从图 5中看出，与普通串行贝叶斯检测僵尸网络相比，MapReduce并行化贝叶斯检测僵尸网络效率较高，并且随着数据量增大，效率优势明显增强。

4 结束语

本文提出了一种利用云环境下的Hadoop机制的MapReduce框架设计与实现贝叶斯分类的僵尸网络检测方法。与已有的僵尸网络检测方法不同的是：它以主机对作为分析对象，提取主机对通信的流量特征，然后将这些特征作为贝叶斯分类算法的输入，基于MapReduce训练生成贝叶斯分类器，用训练好的贝叶斯分类器进行僵尸网络的检测。这种检测方法有较高的检测率并且提高了检测效率。另外，本文在训练形成贝叶斯分类器阶段存在如何选择各特征值的阈值范围的问题，阈值范围的选取影响僵尸网络的检测率，下一步工作将对此另行研究。