大数据分析相关技术

　　第二章 相关技术与理论

　　2.1 引言。

　　本文构建的面向大数据的网络安全分析方法涉及的技术领域主要包括有 spark、hadoop集群构建的方法、异常流量的数据挖掘算法等，以及将这些技术应用到基于大数据的网络安全分析中。同时，在不失一般性的情况下，本文中所涉及的网络异常检测对象采用的是 DDoS攻击，其理论和实验环境可以很方便的移植到其他网络异常检测场景下。因此，对于大数据网络安全分析中相关的关键技术进行系统的描述和分析对于本文后面章节中构建大数据网络安全分析系统具有重要的指导意义。

　　本章节具体内容安排如下：2.2 小节详细介绍了 DDoS 攻击研究相关现状；2.3 小节系统阐述了 hadoop 生态圈和 spark 集群相关背景及其知识；2.4 节系统详细描述了 sparkmllib 库和数据挖掘算法；2.5 小节系统详细的描述了大数据在网络安全领域的应用和优势。

　　2.2 DDoS 攻击检测相关研究现状。

　　由CERT/CC发起的DIStributed-SystemS Intruder Tools workshop于1999年11月份发表了一份报告[16],最早对 DDoS 问题做了全面的阐述，详细分析了分布式系统攻击的原理，并分别针对系统管理员、ISP、以及事件响应小组，从预防、检测、响应三 fangm 依次给出了即时、短期、长期的解决方案。紧接着几个著名的安全组织（如 SANS）和公司（如 eise.SystemS）也给出了防御 noos 攻击的建议[17][18].DHS、NFS 资助几个大学和公司启动了 DETER（defenseTeChnology Experimental Researeh），这个项目的一个研究重点就是防御 DDoS 攻击[19].我国信息产业部在 2005 年公开向产业界招标防御 DDoS 攻击系统的设计方案。另外还有许多公司、高校和科研机构都在从事这方面的研究工作。

　　在攻击研究方面，每种典型的 DDoS 工具（Trin00、TFN、Staeheldraht、TFNZK、Shaft、mstream 等）都被做了详细的分析。这些工作一方面为 DDoS 攻击的防御提供了重要参考， 另一方面也使攻击者能进一步改进攻击工具。

　　在 DDoS 检测方面，Christis Papadopoulos[20]等提出协同阻止并发攻击的方法，通过分布式的检测 DDoS 攻击，使用位于 Internet 边缘网络的组件和其它边缘网络一起对付 DDoS 攻击；在DDos 追踪方面，Savage[21]提出概率包标记方法，通在让路由器对经过的数据包中按一定概率对数据包进行标记，使得被标记的数据包含有它经过的边的信息，这样为追踪攻击源提供了有效的依据；在 DDoS 防御方面。Peng[22]等提出用过滤的方法来有效缓解 DDoS 攻击，使用软件的方式实现基于历史纪录的 IP 过滤方法，这种方法在边界路由器里保留有在网络里曾经出现的所有合法 IP 地址的历史纪录，当边界路由器负载严重时，就通过历史纪录以决定是否允许 IP 包进入。国内的研究机构和个人也做了许多工作。在追踪方面，李德全等采用 IP 追踪的包标记术[23],通过对几种包标记方案的分析，对基本型概率边采样 FMs 方案的修改使得计算量缩减很多。胡小新提出了一种 DDoS 攻击防御体[24],通过在局部范围对 DDoS 攻击进行消除。卢建芝提出了一种配置在源末端网络的 DDos 防御系统[25],通过不断监控网络与其他因特网之间的双向业务流并周期性地与正常业务模型相比较，对攻击进行检测，并通过对可疑流进行限速，使得即使在攻击进行中，也能为合法业务提供很好服务。而随着 DDOS 攻击的频频爆发，越来越多的公司和科研单位加入到 DDoS 攻击防范的研究之中，北京大学计算机科学技术研究所将 DDos 攻击防范产品的研究作为将来的重点研究项目[26].上海市科委将"大规模 DD0s 防御技术"的研究作为 2004 年度信息技术领域重点科技攻关项目[27]从目前的DDoS攻击技术和防御现状来看，功防双方的较力还在继续，DDOS攻击工具仍在不断地升级改进，己有的办法、理论和产品可以产生一些效果，但是还不能消除根本问题。一些产品还没有得到广泛应用，其效果如何很难判断。并且已有的办法和理论大多只是尽量缓解DDoS 攻击，很多防御方法具有单一性，在面对混合型攻击时，就无法施展。DDoS 防御机制还有待进一步的完善和提高。

　　2.2.1 DDoS 攻击概述。

　　Dos 攻击指的是合理向服务器请求资源来占用过于多的服务资源，从而导致合法的用户在请求资源时无法得到服务的响应。该攻击侧重于通过对主机特定漏洞利用攻击导致网络栈失效，机器死机而无法提供正常的服务，dos 攻击一般采取一对一的方式，ddos 攻击手段由 teardrop,land,jolt,icmp,smurf 等。当被攻击的目标 cup,内存或者网络带宽性能较低时它的攻击效果是很明显，但是随着计算机网络技术的迅速发展，计算机 cpu,内存还有带宽各方面都得到了显著的提升，这就使得 dos 攻击的难度增大，此时 ddos 攻击就应运而生，ddos 攻击则是通过很多傀儡机向受害主机发送大量看似合法的网络包，从而造成服务器资源耗尽和网络阻塞而导致拒绝服务。常见的 ddos 攻击 syn flood[28],ack flood ,udp flood ,icmpflood,tcp flood,connection flood 等。

　　在 DDoS 攻击原理图中，DDoS 攻击分为：攻击方、受控制的傀儡计算机和发动攻击的傀儡计算机以及目标系统[29].

　　攻击方把发动攻击的命令发送给控制傀儡计算机。控制傀儡计算机，是指攻击方通过使用不正当的手段来入侵的计算机，同时控制这些被入侵的主机，这些被控制的机器还能控制大量其他的机器。控制傀儡计算机通过本机上特有的服务程序来接受攻击方下达的攻击命令，当被控制的傀儡机器接到攻击指令后，然后把攻击指令发送到发动到直接进行攻击的傀儡主机上。

　　发动攻击的傀儡主机可以直接执行攻击行为。通常情况下，攻击方也会通过一些不正当的手段入侵傀儡主机，并且对其进行控制，只要发动攻击的傀儡集接收到要发动攻击的指令，就会同时开始向目标主机或者系统发起攻击。

　　2.2.2 常见的 DDoS 攻击种类。

　　（1）synflood:这种攻击指的是随机的源主机地址向目的主机发送 SYN 包，目的主机在收到的 SYN ACK 后并不回应，这样，目的主机就为这些源主机建立了大量的连接队列，由于目的主机没有收到 ACK 一直维护着这些队列，造成了计算机资源的大量消耗因此不能向正常请求提供服务。

　　（2）Smurf:该攻击向一个子网的广播地址发一个带有特定请求（如 ICMP 回应请求）的包，并且将源地址伪装成想要攻击的主机地址。子网上所有主机都回应广播包而向被攻击主机发包，使该主机受到攻击。

　　（3）Land-based:攻击者将一个包的源地址和目的地址都设置为目标主机的地址，然后将该包通过 IP 欺骗的方式发送给被攻击主机，这种包可以造成被攻击主机因为试图与自己建立连接而陷入死循环，从而很大程度地降低了系统性能。

　　（4）Ping of Death:根据 TCP/IP 的规范，一个包的长度最大为 65536 字节，当一个主机收到了长度大于 65536 字节的包时，就是受到了 Ping of Death 攻击，该攻击会造成主机的宕机。

　　（5）Teardrop:IP 数据包在网络传递时，数据包可以分成更小的片段。攻击者可以通过发送两段（或者更多）数据包来实现 TearDrop 攻击。第一个包的偏移量为 0,长度为 N,第二个包的偏移量小于 N.为了合并这些数据段，TCP/IP 堆栈会分配超乎寻常的巨大资源，从而造成系统资源的缺乏甚至机器的重新启动。

　　（6） PingSweep:使用 ICMP Echo 轮询多个主机。

　　（7）Pingflood: 该攻击在短时间内向目的主机发送大量 ping 包，造成网络堵塞或主机资源耗尽。

　　2.3 大数据分析相关技术。

　　2.3.1 Hadoop 生态圈。

　　Hadoop 是一个能够对大量数据进行分布式处理的软件框架。为了完成这种计算功能，hadoop 有三个基本组件，分别叫 hdfs,yarn 和 mapreduce.

　　（1）MapReduceMapReduce 是一套软件框架，包括 Map（映射）和 Reduce（化简）两个阶段，这两个阶段可以进行海量数据集的切分、任务分解和结果汇总，从而完成海量数据的并行处理。如图2.2 所示，MapReduce 采用的是先分割后合并的数据处理方式作为它的工作原理。Map 指的是"分解",把海量的数据集切分成了许多若干部分，分散给多台主机并行处理；Reduce 就是指"合并",将各台处理器处理后的结果进行合并汇总从而得到最终处理分析结果。

　　mapreduce 由以下几个组件构成：

　　1）用户程序：计算任务的发送和接受的主要组成部分。

　　2）Master:master 节点给 worker 节点安排计算任务。

　　3）Worker:集群中做实际处理的节点。

　　（a）输入切分数据片（input split）：
　　
　　mapreduce 会在 map 计算之前根据输入文件计算输入分片（input split），每个输入分片（input split）针对一个 map 任务，输入分片（input split）存储的是一个分片长度和一个记录数据的位置的数组而并不是数据本身，输入切分的分片（input split）往往和 hdfs 的 block（块）关系很密切，假如我们设定 hdfs 的存储块的大小为 64mb,如果我们输入三个文件，这三个文件大小分别为 127mb、3mb、65mb 和，那么 mapreduce 就会把 3mb 文件做为一个输入分片（input split），65mb 则是两个输入分片（input split）而 127mb 也是两个输入分片（input split），换句话说我们如果在 map 计算前做输入分片调整，例如合并小文件，那么就会有 5 个 map 任务将执行，由于每个执行 map 函数的数据大小不均，所以这也是 mapreduce 优化计算的一个关键点所在。

　　（b）map 阶段：

　　指的是被程序员编写好的 map 函数，所以 map 函数效率相对好控制，而且一般 map 操作都是本地化操作也就是在数据存储节点上进行。

　　（c）combiner 阶段：

　　combiner 其实相当于一种 reduce 操作，因此我们看见 WordCount 类里是用 reduce 进行加载的。Combiner 就是指一个类似本地化的 reduce 操作，作为 map 运算的后续操作，它的作用是在 map 计算出中间文件之前做简单的合并重复 key 值的操作，例如我们对文件里的单词频率做统计，map 计算时候如果碰到一个 hadoop 的单词就会记录为 1,但是这篇文章里 hadoop可能会出现 n 多次，那么 map 输出文件冗余就会很多，因此在 reduce 计算前对相同的 key 做一个合并操作，那么文件会变小，这样就提高了宽带的传输效率，毕竟 hadoop 计算力宽带资源往往是计算的瓶颈也是最为宝贵的资源，但是 combiner 操作是有风险的，使用它的原则是combiner 的输入不会影响到 reduce 计算的最终输入，例如：在求总数，最大值，最小值时可以使用 combiner,但是做平均值计算使用 combiner 的话，最终的 reduce 计算结果就会出错。

　　（d）shuffle 阶段：

　　shuffle 就是将 map 的输出结果作为 reduce 的输入的数据的过程，这个是 mapreduce 优化的重点地方。这里我不讲怎么优化 shuffle 阶段，讲讲 shuffle 阶段的原理，因为大部分的书籍里都没讲清楚 shuffle 阶段。Shuffle 一开始就是 map 阶段做输出操作， mapreduce 计算的一般都是海量的数据，map 在输出时并不能将所有的文件都放到内存中进行操作，所以 map 写入磁盘的过程十分的复杂，更何况 map 输出时候要对结果进行排序，内存在开销方面花费是巨大的，map 在做输出时候，map 会在内存中打开一个类似环状内存缓冲区，该缓冲区是专门用来做数据输出的，默认大小是 100mb,这个缓冲区在配置文件里被设定了一个默认是 0.80阀值，（这个大小和阀值都是可以在配置文件里进行配置的），同时 map 也会启动守护线程在输出操作时，当缓冲区的内存达到了阀值 80%,那么守护线程就会在磁盘上写入内容，我们称这个操作过程叫 spill,其他磁盘的数据可以写进其他 20%内存，写入磁盘和内存这两个操作是互不干扰的，如果撑满了缓存区，那么写入内存的操作就会被 map 阻塞，让写入磁盘操作完成后再继续执行写入内存操作，前面我讲到写入磁盘之前有一个排序操作，这个操作在是在写入磁盘时候进行，而不是在读入内存时候进行的，如果我们当时定义了一个 combiner函数，那么 combiner 操作就会在排序前执行。

　　spill 操作就是在写入磁盘操作时候，写一个溢出文件。这个过程里还会有一个 Partitioner操作，对于这个操作很多人都很迷糊，其实 Partitioner 操作和 map 阶段的输入分片（Input split）很像，一个 Partitioner 对应一个 reduce,假如 mapreduce 只有一个 reduce 操作，那么 Partitioner就只有一个，如果我们有多个 reduce 操作，那么 Partitioner 对应的就会有多个，Partitioner 因此就是 reduce 的输入分片，这个程序员可以编程控制，是根据 key 和 value 的值，根据实际业务或者更好的 reduce 负载均衡要求而进行，这也是提高 reduce 效率的关键。reduce 阶段一般合并 map 输出的文件，相对应的 map 输出文件会被 Partitioner 找到，接着进行复制操作，几个复制线程也会在复制操作时开启，这些线程通常默认是 5 个，复制线程的个数也可以被程序员在配置文件时更改，它的复制过程和 map 写入磁盘过程类似，也有阀值和内存大小，也可以在配置文件里配置阈值文件，可以直接使用 reduce 中的 tasktracker 的内存大小作为内存大小，合并文件和排序操作也会在 reduce 时进行，reduce 操作也会在这些操作结束后进行。

　　（e）reduce 阶段：和 map 函数一样也是程序员编写的，最终结果是存储在 hdfs 上的。

　　（2）HDFS 分布式文件系统。

　　hadoop 的分布式文件系统 hdfs 是一个用于在多个节点之间存储大型数据集的框架。HDFS主要由三个主要部分组成，namenode,datanodes 和 client.Namenode 存储 metadata 比如数据块的地址和文件的名字。Datanode 存储实际的数据。Client 是用来连接文件系统的。

　　（3）YARN:集群的管理和资源调度YARN（Yet Another Resource Negotiator,另一种资源协调者）是一种新的 Hadoop 资源管理器，它是一个可为上层应用提供统一资源管理调度的通用资源管理系统，它的引入为集群在利用率、资源统一管理和数据共享等方面带来了巨大好处。Yarn 主要由以下几个部分组成：

　　Client:资源管理的接口，用来提交 mapreduce 程序到集群。

　　Resource manager:管理资源。

　　Node manager:管理单节点的资源。

　　Application manager:管理但应用的任务。

　　Container:单节点的资源收集器。

　　同时，Hadoop 框架也存在着很大的缺点，对于低延迟数据访问无法适应、对同时存储大量小文件也无法做到、对多用户写入和修改文件也不支持[30].围绕 Hadoop 而开发的开源项目为了弥补 Hadoop 这些方面存在的缺陷正在逐渐兴起，例如：① Hive: 是一种按列式存储的分布式数据仓库，这种数据仓库可以管理存储在 HDFS 中的数据，并提供基于类似于 SQL 查询语言用来查询数据；② Zookeeper:使一种可用性高的分布式的协调服务，它可以提供分布式锁之类的基本服务，这种服务可以用于构建分布式应用[31];③ HBase:它是一个分布式按列存储数据库，HDFS 作为它的底层存储，同时也支持MapReduce 的批量计算与点查询。

　　2.3.2 SPARK 集群。

　　（1）简介。

　　Spark 使 UC berkeley AMP lab 所开发的类似于 hadoop 的 mapreduce 的分布式通用计算框架。Spark 拥有所有 mapreduce 的优点，但是有一点不同的是 spark 的中间计算结果可以保存在内存中，所以不需要从 hdfs 中读写文件，所以 spark 对于机器学习和数据挖掘等迭代式的算法效率非常高[32].

　　Spark 可以很容易与 yarn 结合，并且可以直接从 hdfs,hbase 等文件系统中读取数据，它同时也可以架设在 hadoop 之上，近年来，spark 技术发展十分迅猛，而且这种计算框架也比hadoop 更加灵活方面，它一方面建设了数据处理的延时，另一方面提高了性能和使用等灵活性。

　　Spark 的核心部分叫做 RDD（弹性分布式数据集）。Spark SQL、Spark Streaming、MLlib、GraphX、Spark R 等核心组件分别解决了许多的大数据问题，其完美的框架日受欢迎。其相应的生态环境包括 zepplin 等可视化方面，正日益壮大。大型公司争相实用 spark 来代替原有hadoop 上相应的功能模块。Spark 读写过程不像 hadoop 溢出写入磁盘，都是基于内存，因此速度很快。另外 DAG 作业调度系统的宽窄依赖让 Spark 速度提高。

　　（2）Spark 核心组成。

　　（a）RDD.

　　是弹性分布式数据集，完全弹性的，如果数据丢失一部分还可以重建。有自动容错、位置感知调度和可伸缩性，通过数据检查点和记录数据更新金象容错性检查。通过SparkContext.textFile（）加载文件变成 RDD,然后通过 transformation 构建新的 RDD,通过 action将 RDD 存储到外部系统。

　　RDD 使用延迟加载，也就是懒加载，只有当用到的时候才加载数据。如果加载存储所有的中间过程会浪费空间。因此要延迟加载。一旦 spark 看到整个变换链，他可以计算仅需的结果数据，如果下面的函数不需要数据那么数据也不会再加载。转换 RDD 是惰性的，只有在动作中才可以使用它们。

　　Spark 分为 driver 和 executor,driver 提交作业，executor 是 application 早 worknode 上的进程，运行 task,driver 对应为 sparkcontext.Spark 的 RDD 操作有 transformation、action.

　　Transformation 对 RDD 进行依赖包装，RDD 所对应的依赖都进行 DAG 的构建并保存，在worknode 挂掉之后除了通过备份恢复还可以通过元数据对其保存的依赖再计算一次得到。当作业提交也就是调用 runJob 时，spark 会根据 RDD 构建 DAG 图，提交给 DAGScheduler,这个 DAGScheduler 是在 SparkContext 创建时一同初始化的，他会对作业进行调度处理。当依赖图构建好以后，从 action 开始进行解析，每一个操作作为一个 task,每遇到 shuffle 就切割成为一个 taskSet,并把数据输出到磁盘，如果不是 shuffle 数据还在内存中存储。就这样再往前推进，直到没有算子，然后运行从前面开始，如果没有 action 的算子在这里不会执行，直到遇到 action 为止才开始运行，这就形成了 spark 的懒加载，taskset 提交给 TaskSheduler 生成TaskSetManager 并且提交给 Executor 运行，运行结束后反馈给 DAGScheduler 完成一个taskSet,之后再提交下一个，当 TaskSet 运行失败时就返回 DAGScheduler 并重新再次创建。

　　一个 job 里面可能有多个 TaskSet,一个 application 可能包含多个 job.

　　（b）Spark Streaming.

　　通过对 kafka 数据读取，将 Stream 数据分成小的时间片段（几秒），以类似 batch 批处理的方式来处理这一部分小数据，每个时间片生成一个 RDD,有高效的容错性，对小批量数据可以兼容批量实时数据处理的逻辑算法，用一些历史数据和实时数据联合进行分析，比如分类算法等。也可以对小批量的 stream 进行 mapreduce、join 等操作，而保证其实时性。针对数据流时间要求不到毫秒级的工程性问题都可以。

　　Spark Streaming 也有一个 StreamingContext,其核心是 DStream,是通过以组时间序列上的连续 RDD 来组成的，包含一个有 Time 作为 key、RDD 作为 value 的结构体，每一个 RDD都包含特定时间间隔的数据流，可以通过 persist 将其持久化。在接受不断的数据流后，在blockGenerator 中维护一个队列，将流数据放到队列中，等处理时间间隔到来后将其中的所有数据合并成为一个 RDD（这一间隔中的数据）。其作业提交和 spark 相似，只不过在提交时拿到 DStream 内部的 RDD 并产生 Job 提交，RDD 在 action 触发之后，将 job 提交给 jobManager中的 JobQueue,又 jobScheduler 调度，JobScheduler 将 job 提交到 spark 的 job 调度器，然后将 job 转换成为大量的任务分发给 spark 集群执行[26].Job 从 outputStream 中生成的，然后触发反向回溯执行 DStreamDAG.在流数据处理的过程中，一般节点失效的处理比离线数据要复杂。Spark streaming 在 1.3 之后可以周期性的将 DStream 写入 HDFS,同时将 offset 也进行存储，避免写到 zk.一旦主节点失效，会通过 checkpoint 的方式读取之前的数据。当 worknode节点失效，如果 HDFS 或文件作为输入源那 Spark 会根据依赖关系重新计算数据，如果是基于 Kafka、Flume 等网络数据源 spark 会将手机的数据源在集群中的不同节点进行备份，一旦有一个工作节点失效，系统能够根据另一份还存在的数据重新计算，但是如果接受节点失效会丢失一部分数据，同时接受线程会在其他的节点上重新启动并接受数据。

　　（c） Graphx.

　　主要用于图的计算。核心算法有 PageRank、SVD 奇异矩阵、TriangleConut 等。

　　（d）Spark SQL是 Spark 新推出的交互式大数据 SQL 技术。把 sql 语句翻译成 Spark 上的 RDD 操作可以支持 Hive、Json 等类型的数据。

　　（e） Spark R.

　　通过 R 语言调用 spark,目前不会拥有像 Scala 或者 java 那样广泛的 API,Spark 通过 RDD类提供 Spark API,并且允许用户使用 R 交互式方式在集群中运行任务。同时集成了 MLlib机器学习类库。

　　（f） MLBase从上到下包括了 MLOptimizer（给使用者）、MLI（给算法使用者）、MLlib（给算法开发者）、Spark.也可以直接使用 MLlib.ML Optimizer,一个优化机器学习选择更合适的算法和相关参数的模块，还有 MLI 进行特征抽取和高级 ML 编程 抽象算法实现 API 平台，MLlib分布式机器学习库，可以不断扩充算法。MLRuntime 基于 spark 计算框架，将 Spark 的分布式计算应用到机器学习领域。MLBase 提供了一个简单的声明方法指定机器学习任务，并且动态地选择最优的学习算法。

　　（g） Tachyon.

　　高容错的分布式文件系统。宣称其性能是 HDFS 的 3000 多倍。有类似 java 的接口，也实现了 HDFS 接口，所以 Spark 和 MR 程序不需要任何的修改就可以运行。目前支持 HDFS、S3 等。