5.13Spark编程实例——Log挖掘在线视频

好 下面我们以log挖据

来以一个非常简单的例子

为例来看

这个Spark执行的这样的一个

过程到底是怎么样的

那么它其实很简单

它就是做的操作

就是将数据从文件系统中

先调到内存里面

然后进行一些交互式的查询

我们现在有这样的几台机器

其中有一个是Driver

然后另外几个是Worker

就相当于这是一个控制的部分

那么剩下几个是做工作的部分

那么我们第一个Spark的语句

是这个spark textFile

它会产生一个RDD

从这个文件系统中

hdfs从这个文件系统中

产生这样的一个RDD Lines

这是我们的这个基础的RDD

然后我们从这个Lines里面

我们去过滤

过滤到其中这个

所有还有以ERROR开头的

这样的一些语句

这是一个变换也就是说

我们把Lines把它变换到了

这个ERROR的这个部分

然后我们继续的去做这个分词

把这个ERROR把它来做split

把它做分词

然后生成这个massages

这样一个RDD

那么这样的一个

后面这个messages cache

这样的一个调用

它说明什么

是说我们要尽可能的把这个

massages的这个信息

保留在内存里面

到在我们用完了这个

所以这样以后这个cacheMsgs

它会尽量的保留在内存

下面我们如果要对这个

cacheMsgs做多次查询的话

这个部分系统会尽量的

把它放在内存里

而不需要重新生成

或者是重新排上去读取

第一个就是去查说

在这些这个错误的信息里

包含有FOO的这个消息

到底有多少

然后最后把它的count输出

这个是这样的一个操作

这个count我们刚才讲过

count就是一个Action

那么这个数据

现在这个cacheMsgs它可能

分布在多个机器上

有Worker1 Worker2

Block3分布在不同的机器上

那么这样一个操作

就由Drivr把这样的一个命令

发给不同的Worker

在本地的这个做完了

这样的一些

实际做完了这个filter的

这个操作以后会把结果

返回给Driver

Drivr这时候就

返回这样的一个技术值

也就是说

在这个产生的这个消息里面

含有FOO的这些消息到底有多少个

那么这个每台机器上

实际上还有

保存了这个消息的cache

那么这样的话

这个cache我们刚才因为

已经调用了这个cache

这样的一个东西

那么所以这个消息

现在cacheMsgs

现在实际上是

如果内存量足够的话

它会缓存在内存里面

所以当我们在做下一个

contains bar这样一个

查询的时候

这个查询会直接在这个cache

上面来进行

并且把结果返回给Driver

所以这个性能在1TB的数据

可以在5-7秒内完成

而数据如果在硬盘上的话

需要170秒的时间

才能够完成

所以可以看到说

性能得到了很大的提高

下面这个例子

是一个逻辑回归的

这样一个例子

这是一个机器学习的一个算法

那么这个算法首先也是

从这个文件里面

把数据读出来

然后把它cache

到这个内存里面

然后这个生成一个

随机的一个向量

那么这个向量

后面我们再继续

再去做这个迭代的这个

Map Reduce就是这部分

实际上就是整个机器学习的

这样一个过程

也就是说我们

先有一个随机的

这样的一个线

那么通过这个线

我们算一下说它和

现有的这些点之间的这个关系

通过这样的一个公式

可以算出它的这样一个T度

那么用这个W

去减掉这个T度以后

我们就和这个能够实际分类的

这个向量就更加的接近一些

产生我们要做多个迭代

每次迭代实际上都是

先去做这个MAP

就是把每个点对每个点

我们都去算这个

它和当前的这个W之间的

这个向量之间的

这样的一个点击

这样的一个结果

那么这个结果算完以后

实际上最后生成

这样的一个T度值

那么这个T度值最后

是让这个向量

去减去这个T度值以后

再做下一次的迭代

那么迭代多少次

这个是事先设定的

也就是这个算法

最后也是一个

这个不断迭代以后收敛以后的

一个算法

最后就可以生成

我们真正的逻辑回归的结果

就是这个分类的向量

就可以生成了

那么这个逻辑回归的性能

我们也可以对比一下

如果用Hadoop来做

这样的一个逻辑回归

对同样的数据集来讲

如果要算比如说30个迭代的话

需要将近一个小时

那么我们可以看到

这个在Spark

实际上它需要的时间

就远远少于这样的一个时间

但是我们也可以看到

这样一点就是说

在做这个第一个迭代的时候

实际上Spark的这个时间

还略高于这个Hadoop的时间

我们在这儿也可以看到

这个Spark它的第一个

iteration第一个迭代

它大概需要174秒

而Hadoop它平均来讲

它要花127秒来做一个迭代

那么后续的

因为我们cache了这个数据

所以后续的我们只需要

这个6秒钟来做一个新的迭代

所以这个时间就大大的缩短了