第1期 邵秀丽，等：检测僵尸网络的贝叶斯算法的MapReduce并行化实现 ·31 进行训练，每个属性既要判断是否为僵尸网络又要 判断是否在阈值内，因此每个属性有4个判断条件。 P=P w1_out_normal; 因此，经过MapReduce22的处理，形成24个条件概 P2=P2 w1_out_unnormal; 率分别存储在24个成员变量里，这与MapReduce1 形成的2个成员变量存储的先验概率共同构成完整 /其他5个属性列检测过程同上。 的知识库，可用于检测僵尸网络。 Line++;/统计所在行数 2.3 MapReduce3设计 Context.write(Line,P:P2); Map3接收到的检测数据是被Hadoop处理形成 的(Key,Value〉对，形式为（该行起始位置相对于文 经过Map3把分块的每行信息都处理成以 件起始位置的偏移量，文本文件中的一行信息〉的 〈Key5,Value5)形式的等待整体处理的中间文件输 信息。MapReduce.3要对6列属性全部检测，需用到 出，MapReduce框架将每个Map3输出的中间文件 Value的6个属性列。所以Map3将每行Value数据 结果按照Ky值（数据所在行数）进行分组后发送 按空格分割成字符串数组，取出数组的第3~8项， 给Reduce3。 分别为T℃P数据流、时间间隔平均值、时间间隔变 Reduce3.接收到的信息为<Key5,Value5>。Re- 化、数据包字节数、数据包个数平均值、持续时间平 duce5的任务是逐行比较网络数据的P(bId)和 均值。判断6个属性列的值是否在各自阈值内，若 P(nld)的大小。若P(nld)>P(bld),判断该行 在阈值内，分别利用存储条件概率的成员变量wi_in 网络数据为正常网络数据：否则为僵尸网络数据。 nomal、wi_in_unnormal计算后验概率；若在阈值外， Reduce.3伪代码如下所示。 分别利用存储条件概率的成员变量wi_out_nomal、 输入：Text、Text。 wi_out_unnormal计算后验概率。并将每行网络数 输出：Text、Text。 据的正常网络后验概率P(nId)和僵尸网络后验 reduce(Key,Value) 概率P(bId)一起输出。输出结果(Key5,Values5) StringTokenizer itr new StringTokenizer(value. 对的形式为（数据所在行数，P(nId)P(bId)〉。 toString()); Map3过程伪代码如下。 String[temp=new String[2] 输入：Object、Text。 While(itr.hasMoreTokens()) 输出：Text、Texto {temp[]=itr.nextToken();/正常网络后 map(Key,Value) 验概率与僵尸网络后验概率 StringTokenizer itr new StringTokenizer(value. i++: toString()); String[temp=new String[9]; If(Float.parseFloat(temp[0])>Float.parseFloat While(itr.hasMoreTokens()) (temp[1]))/比较 {temp[i]=itr.nextToken()://属性字符串数组 Context..write(Key,“正常网络”)：/判断 i++; Else Context.write(Key,“僵尸网络”)：//判断 P =sum_yes_p;P2=sum_no_p; If Float.parsefloat temp [2])>140&&Float. 3实验结果与分析 parsefloat(temp[2])<150) 本文实验中的被测网络环境为某校园网中一个 P=P w1_in_nomal; 子网的流量，该子网内主机约200台，白天的网络流 P,=P,wl_in_unnormal; 量为150~200Mbps。实验采集了某天数据，为测试 本文提出并行化的算法性能，分别使用了2个不同 Else if(Float.parsefloat temp[2])<140 ll Float. 时间段的数据集D,和D2。D,解析后的文本文件 parsefloat(temp[2])>150) 1.6GB,TCP数据包个数23631638。D2解析后的进行训练，每个属性既要判断是否为僵尸网络又要 判断是否在阈值内，因此每个属性有 ４ 个判断条件。 因此，经过 ＭａｐＲｅｄｕｃｅ２ 的处理，形成 ２４ 个条件概 率分别存储在 ２４ 个成员变量里，这与 ＭａｐＲｅｄｕｃｅ１ 形成的 ２ 个成员变量存储的先验概率共同构成完整 的知识库，可用于检测僵尸网络。 ２．３ ＭａｐＲｅｄｕｃｅ３ 设计 Ｍａｐ３ 接收到的检测数据是被 Ｈａｄｏｏｐ 处理形成 的〈Ｋｅｙ，Ｖａｌｕｅ〉对，形式为〈该行起始位置相对于文 件起始位置的偏移量， 文本文件中的一行信息〉的 信息。 ＭａｐＲｅｄｕｃｅ３ 要对 ６ 列属性全部检测，需用到 Ｖａｌｕｅ 的 ６ 个属性列。 所以 Ｍａｐ３ 将每行 Ｖａｌｕｅ 数据 按空格分割成字符串数组，取出数组的第 ３ ～ ８ 项， 分别为 ＴＣＰ 数据流、时间间隔平均值、时间间隔变 化、数据包字节数、数据包个数平均值、持续时间平 均值。 判断 ６ 个属性列的值是否在各自阈值内，若 在阈值内，分别利用存储条件概率的成员变量 ｗｉ＿ｉｎ ＿ｎｏｍａｌ、ｗｉ＿ｉｎ＿ｕｎｎｏｒｍａｌ 计算后验概率；若在阈值外， 分别利用存储条件概率的成员变量 ｗｉ＿ｏｕｔ＿ｎｏｍａｌ、 ｗｉ＿ｏｕｔ＿ｕｎｎｏｒｍａｌ 计算后验概率。 并将每行网络数 据的正常网络后验概率 Ｐ（ｎ ｜ ｄ） 和僵尸网络后验 概率 Ｐ（ｂ ｜ ｄ） 一起输出。 输出结果〈Ｋｅｙ５，Ｖａｌｕｅ５〉 对的形式为〈数据所在行数， Ｐ（ｎ ｜ ｄ） Ｐ（ｂ ｜ ｄ） 〉。 Ｍａｐ３ 过程伪代码如下。 输入：Ｏｂｊｅｃｔ、Ｔｅｘｔ。 输出：Ｔｅｘｔ、Ｔｅｘｔ。 ｍａｐ（Ｋｅｙ，Ｖａｌｕｅ） ｛ＳｔｒｉｎｇＴｏｋｅｎｉｚｅｒ ｉｔｒ ＝ ｎｅｗ ＳｔｒｉｎｇＴｏｋｅｎｉｚｅｒ（ ｖａｌｕｅ． ｔｏＳｔｒｉｎｇ（））； Ｓｔｒｉｎｇ［］ ｔｅｍｐ ＝ ｎｅｗ Ｓｔｒｉｎｇ［９］； Ｗｈｉｌｅ（ｉｔｒ．ｈａｓＭｏｒｅＴｏｋｅｎｓ（）） ｛ｔｅｍｐ［ｉ］ ＝ｉｔｒ．ｎｅｘｔＴｏｋｅｎ（）；／ ／ 属性字符串数组 ｉ＋＋； ｝ Ｐ１ ＝ ｓｕｍ＿ｙｅｓ＿ｐ；Ｐ２ ＝ ｓｕｍ＿ｎｏ＿ｐ； Ｉｆ （ Ｆｌｏａｔ． ｐａｒｓｅｆｌｏａｔ （ ｔｅｍｐ ［ ２ ］） ＞ １４０＆＆Ｆｌｏａｔ． ｐａｒｓｅｆｌｏａｔ（ｔｅｍｐ［２］）＜１５０） ｛ Ｐ１ ＝Ｐ１∗ｗ１＿ｉｎ＿ｎｏｍａｌ； Ｐ２ ＝Ｐ２∗ｗ１＿ｉｎ＿ｕｎｎｏｒｍａｌ； ｝ Ｅｌｓｅ ｉｆ（Ｆｌｏａｔ． ｐａｒｓｅｆｌｏａｔ（ ｔｅｍｐ［２］） ＜１４０‖Ｆｌｏａｔ． ｐａｒｓｅｆｌｏａｔ（ｔｅｍｐ［２］）＞１５０） ｛ Ｐ１ ＝Ｐ１∗ｗ１＿ｏｕｔ＿ｎｏｒｍａｌ； Ｐ２ ＝Ｐ２∗ｗ１＿ｏｕｔ＿ｕｎｎｏｒｍａｌ； ｝ ／ ／ 其他 ５ 个属性列检测过程同上。 Ｌｉｎｅ＋＋； ／ ／ 统计所在行数 Ｃｏｎｔｅｘｔ．ｗｒｉｔｅ（Ｌｉｎｅ，Ｐ１ Ｐ２ ）； ｝ 经过 Ｍａｐ３ 把 分 块 的 每 行 信 息 都 处 理 成 以 〈Ｋｅｙ５，Ｖａｌｕｅ５〉形式的等待整体处理的中间文件输 出，ＭａｐＲｅｄｕｃｅ 框架将每个 Ｍａｐ３ 输出的中间文件 结果按照 Ｋｅｙ 值（数据所在行数）进行分组后发送 给 Ｒｅｄｕｃｅ３。 Ｒｅｄｕｃｅ３ 接收到的信息为＜Ｋｅｙ５，Ｖａｌｕｅ５＞。 Ｒｅ⁃ ｄｕｃｅ５ 的任务是逐行比较网络数据的 Ｐ（ｂ ｜ ｄ） 和 Ｐ（ｎ ｜ ｄ） 的大小。 若 Ｐ（ｎ ｜ ｄ） ＞ Ｐ（ｂ ｜ ｄ） ，判断该行 网络数据为正常网络数据；否则为僵尸网络数据。 Ｒｅｄｕｃｅ３ 伪代码如下所示。 输入：Ｔｅｘｔ、Ｔｅｘｔ。 输出：Ｔｅｘｔ、Ｔｅｘｔ。 ｒｅｄｕｃｅ（Ｋｅｙ，Ｖａｌｕｅ） ｛ＳｔｒｉｎｇＴｏｋｅｎｉｚｅｒ ｉｔｒ ＝ ｎｅｗ ＳｔｒｉｎｇＴｏｋｅｎｉｚｅｒ（ ｖａｌｕｅ． ｔｏＳｔｒｉｎｇ（））； Ｓｔｒｉｎｇ［］ ｔｅｍｐ ＝ ｎｅｗ Ｓｔｒｉｎｇ［２］； Ｗｈｉｌｅ（ｉｔｒ．ｈａｓＭｏｒｅＴｏｋｅｎｓ（）） ｛ ｔｅｍｐ［ ｉ］ ＝ ｉｔｒ． ｎｅｘｔＴｏｋｅｎ（ ）； ／ ／ 正常网络后 验概率与僵尸网络后验概率 ｉ＋＋； ｝ Ｉｆ（Ｆｌｏａｔ． ｐａｒｓｅＦｌｏａｔ （ ｔｅｍｐ ［ ０］） ＞Ｆｌｏａｔ． ｐａｒｓｅＦｌｏａｔ （ｔｅｍｐ［１］）） ／ ／ 比较 Ｃｏｎｔｅｘｔ．ｗｒｉｔｅ（Ｋｅｙ，“正常网络”）； ／ ／ 判断 Ｅｌｓｅ Ｃｏｎｔｅｘｔ．ｗｒｉｔｅ（Ｋｅｙ，“僵尸网络”）； ／ ／ 判断 ｝ ３ 实验结果与分析 本文实验中的被测网络环境为某校园网中一个 子网的流量，该子网内主机约 ２００ 台，白天的网络流 量为 １５０～２００ Ｍｂｐｓ。 实验采集了某天数据，为测试 本文提出并行化的算法性能，分别使用了 ２ 个不同 时间段的数据集 Ｄ１ 和 Ｄ２ 。 Ｄ１ 解析后的文本文件 １．６ ＧＢ， ＴＣＰ 数据包个数 ２３ ６３１ ６３８。 Ｄ２ 解析后的 第 １ 期 邵秀丽，等：检测僵尸网络的贝叶斯算法的 ＭａｐＲｅｄｕｃｅ 并行化实现 ·３１·