Video在线视频

好

我们刚刚介绍了

这个Spark的好处

它可以将我们原有的这种

在分布式文件系统里面的文件

变成了一个一个的这种RDD

在内存中或者在硬盘中存在

同时具有很好的partition

和data locality的特性

那么我们下面就给一些例子

我们来看看Spark的能力

到底是什么样的

那么这些例子希望能够

让同学们感受一下

Spark它在你自己的

这个数据分析或者数据处理里面

能做到什么样的这个事情

后面我们还会再教大家

在真实的一个集群上

到底怎么样去运行一个

Spark的程序

那么在那个之前

我们先看看它的例子是什么样的

第一个例子

也是Spark它强调自己的

这样一个特性的一个点

就是说交互性的操作

我们来看

怎么样通过交互性的操作

来对日志进行挖掘

那在这个例子里面

这些日志实际上是服务器记录的

一些关于系统的一些这个记录

那么它是以一行一行打印出来的

每个日志可能还打印了时间

错误的原因

或者是一些详细的信息

那么我来看

对这样的数据怎么样挖掘

那我相信在我们同学当中

肯定会有一些负责服务器的维护

或者是负责运营的

他们会经常接触这样的数据

那么在传统的方法里面

可能你在单机上或者在map-reduce里面

对这样的数据进行挖掘

我们来看在Spark

底下这样的数据

怎么样得到有效的一个处理

那么这是我们Spark的

一个集群

我们有这么几台机器

有一些worker

这worker

会负责进行真实的

这个代码的一个并行

那么我们现在来看

这一段Spark的代码

它是以scala去写的

大家可以忽略具体的语法

而且看它的逻辑

我们看它第一行

实际上是将数据读入起来

从分布式文件系统HDFS里面

把某一个这个系统运行的

日志的数据给读入起来

读入起来之后

它做了一个transform

这个transform

是进行一个过滤filter

那么它过滤了什么呢

就是说

这一行要以error打头

以error打头的这些行

就被我们过滤了出来

过滤出来之后

生成了一个新的RDD

叫做errors

那么它进一步又做了一个变换

实际上是进行一个切分

切分是以这个tab键

这样一个tab的空行

进行切分

然后把他的这个两个

把它的这个两列给取了出来

放到了message里面

然后对messge

又做了一个cache的操作

把它变成一个在内存中的RDD

那么这是这一系列的这个工作

做了这样一系列的事情

我们来看从文件读出得到了

base RDD

然后base RDD

经过了一系列的

transform

形成了一系列这种中间的RDD

然后有了这个依赖的关系

血缘的关系

然后最后我们进行了一个

这个控制

将这个RDD进行了内存化

那么内存化的数据

内存化的RDD可以在

后面得到反复的运行

迭代的运行

或者是交互的操作

那么它不再需要反复的

实际的计算

而是可以直接从内存中

读到这部分数据

那我们来看

最后用户

他需要做一些什么样的操作呢

他第一步做了一个操作

去在这个缓存的RDD里面去查找

包含某一个关键词的这个行数

然后把它的行数用通过count

统计出来

显示到用户的这个磁盘上

那我们知道根据之前的介绍

这个操作是一个action

那么它会触发之前的这样一些

transform和他的控制

真正的去得到一个运行

它是一个并行的操作

最终会让这些操作在各个worker上

得到一个运行

我们来看

它是怎么样进行一个运行的呢

我们可以看真实的数据

实际上分布在各个worker上

因为是从HDFS

里面读取出来的

那么它照顾到数据的locality

那么在本地就近的

这样一个原则底下

形成了这些RDD的partition

那么任务会通过用户提交到driver

driver又提交到了各个worker上

那么这些worker

知道用户要做一些什么样的

transform和action

那么这些结果最终会被返回到driver

在用户的屏幕上显示出来

那么最终的结果

cachedMsgs

会被cache到各个worker上

那么这些worker在后续的执行当中

可以被反复的调用

那么这是整个执行的过程

它的一个流程是这样的

那么有了这样一个

cached data之后

我们刚刚查过一个关键词

假设我们要换一个关键词

从foo变成了bar之后

实际上这个过程

就不需要再进行了

它只需要在原来的cache里面

去运行这样一个任务

就能达到刚刚完整的

这样一个计算的目标

所以我们发现

有效的支持了迭代化的计算

不管你迭代多少次

只要中间我已经考虑到

要做一个开始操作之后

那么我就不需要反复再去运行

这样一个结果

那么可以支持很好的交互

那么在用户处理这样的

数据的时候

其实它可能会经常需要

去变换它的这个搜索的关键词

或者做一些其他的尝试

那么spark对这两种模式

都有一个很好的支撑

在这个spark的工程师

在开发spark的过程当中

实际上他们也对wikipedia的

这个性能进行了一个验证

那么实际上发现它要很好的

比reduce得到了一个提升

尤其是针对这种迭代和交互性的

这种查询

那我们可以对用spark来对

这种运行的日志进行一个查询

一个简单的任务

那我们再来看spark

还有一些什么样的其他能力

它的其他能力

就包括对map-reduce

的一个支持

也就是说

我们可以用spark里面的transform

和action

完整的把reduce

给模拟出来

我们来看它是怎么样做得到的

那么两种模拟办法

第一个就是在我们这页PPT

上面这一行写到的

首先把数据读取起来

经过flatMap

flatMap实际上它是把每一行

记录通过一个变化

在这里面叫做myMapFunc

那么来进行了一次map

map完了之后

形成了一个新的RDD

那么这个RDD

我们在对它进行一个针对Key的

一个group

也就是会把相同的这个Key对应的

不同的value

整合成一个列表

形成一个新的数据集groupByKey

这个transformation

达到了这样一个目标

最后再去执行真正的这个reduce

map-reduce中的reduce

那么reduce在Hodoop中

是怎么执行的呢

它又进一步通过一个transformation

map来进行

那我们来看这个map函数

怎么写的呢

那么这个函数是这样的

它把一个Key和values的这样

一个list变成了

reduce出来的一个结果

那在reduce里面怎么去写

它的输入呢

这个reduce的输入

正好就是我们给到的Key和value的列表

那么大家回想一下

正好和map-reduce

里面的reduce

需要的输入是一致的

所以经过这样三步transformation

实际上spark

可以完整的模拟出

map-reduce

那么对map-reduce的模拟

其实还包含其他的方法

比如说我们可以通过flatMap

之后再经过reduceByKey

然后再经过map来做到

那么由于spark它提供了

很多的transformation的组合

以及很多的这个action的动作

实际上你要想模拟map-reduce

还有其他的方法

尤其是当你在中间插入一些

cache的过程之后

实际上会大大的改进

原来的这种

map-reduce的性能

就是大家可以在后面的实战当中

去尝试的

那么再给一个例子

我们在map-reduce

当中通过word count

这个例子告诉大家怎么样用

map-reduce

去把一段文档中的

每个单词出现的次数

给统计出来

那我们看在spark当中

我们怎么样来做这件事

当然一个最简单办法

就是利用我们刚刚说到的

map-reduce

既然我们能够模拟

map-reduce

那么我们就可以模拟

map-reduce

怎么样去做word count

假设我们不去直接模拟

map-reduce

我们其实用spark的一些

transformation的操作

也可以很容易做到这件事

第一行

我们从HDFS把原始的数据

读入起来

得到了一个RDD Lines

然后第二行我们把lines

做了一个flatMap

对它的每一行做一个操作

这个操作是把这一行

进行一个分割

那么这个分割的依据是它的空格

所以我们可以把这一行根据空格

分割成若干的单词

那么reduce是做什么呢

reduce我们就是把每个

把这个reduceByKey

就是把每个单词它的这样一个value

的一个输出进行一个叠加

然后最终得到需要的结果

我们可以把这个结果输出到屏幕

也可以直接把这个结果输出到

分布式文件系统

这是reduce做的一个工作

那么在这里面

其实大家可以看到我们用到了scala的

一些这个对函数的一个简写

大家如果在使用spark的时候

需要去用到这样比较灵活的

这种写法的时候

可能大家还需要去参考一下

scala的一些语法手册

那么你们可以写出类似于

这样的一种比较简洁的

这种代码的写法

那么性能

我们知道

这个spark

非常适合用来做迭代

用来做这种交互的行为

这种数据的挖掘

那么一个比较典型的

迭代化的操作就是

logistic regression

就这种逻辑回归

它可以对二分类的

这个问题进行不断的

这样一个迭代

然后得到数据的一个划分

那么这种任务如果大家Hadoop上

和我们之前讨论过的一样

它需要反复的把数据

从分布式文件系统读起

然后操作

然后再写入磁盘

那么这个过程实际上是很慢的

在spark当中

实际上我就可以把中间结果

就是每一轮的这个分类的结果

进行中间结果的一个缓存

那么可以非常有效的降低这样

一个数据处理的时间

当迭代数在增加的时候

我们发现这个性能的差异就会

变得非常的明显

那么spark非常的占优

所以如果大家在后面遇到你的

数据挖掘任务

需要有这种迭代或者是交互的

特征的时候

大家可以去尝试一下

尤其是当你的迭代次数

非常高的时候

大家可以尝试一下spark

带来的这样性能的一个提升

spark有很多它的这个优势

它可以有很好的数据依赖性

把数据处理的任务

分解成了变换和这个action

以及对内存化缓存的一个支持

那么spark可以有效的支持

迭代化的数据处理任务

以及交互的这种数据分析的

这样一个过程

看起来它有很多的优势

但实际上在今天工业界主导的

这个大数据的这样一个行业底下

spark依然不是主流的这样

一个数据分析平台

至少在目前我们看来很多的公司

依然以hadoop

和map-reduce为主

作为他们业务的主要的

这样支撑的基础的流程

那他的原因主要是在

这样的一些企业里面

他们的业务相对是比较稳定的

那么也比较固化

他们需要一个能够批量

能够稳定去执行他们的任务的

这样一个过程

而spark它面向的

环境主要是迭代化

主要是探索性的这样一些项目

比如说某个数据工程师

他需要用很大的这样一个集群

来进行一个科研任务

这样的一个环境

所以spark和hadoop

它依然有自己的

这样一个适用的环境

但是不管怎么说

spark它作为一个新的

这样一个大数据的处理平台

它的生态环境在不断的完善

大家可以看到Databricks

推出了spark之后

现在也支持了spark的

这种流式的数据处理

以spark的sql

spark的机器学习的这样一个框架

以及spark的这种图的

挖掘的系统

那么它的生态在不断地建立起来

那么它的一些不稳定

对内存的使用

有时候不是特别安全的

这样一些问题

也在不断地得到解决

所以我们相信未来这种处理框架

也会慢慢的得到了更多人的使用

或者说是特定的行业

得到一个更大效果的发挥