1.Spark Streaming另类实验与 Spark Streaming本质解析

1 Spark源码定制选择从Spark Streaming入手 

咱们从第一课就选择Spark子框架中的SparkStreaming。

那么，咱们为何要选择从SparkStreaming入手开始咱们的Spark源码版本定制之路？

有下面几个方面的理由：

1）Spark大背景

Spark 最开始没有咱们今天看到的Spark Streaming、GraphX、Machine Learning、Spark SQL和Spark R等相关子框架内容，最开始就只有很原始的Spark Core。咱们要作Spark源码定制，作本身的发行版本，以SparkStreaming为切入点，Spark Streaming自己是 Spark Core上的一个子框架，因此咱们透过一个子框架的完全研究，确定能够精通Spark力量的源泉和全部问题的解决之道；

2）为何不选Spark SQL？

咱们知道，Spark有不少子框架，如今除了基于Spark Core编程以外，用得最多的就是SparkSQL。Spark SQL因为涉及了太多的SQL语法细节的解析或者说优化，其实这些解析或优化，对于咱们集 中精力去研究Spark而言，它是一件重要的事情，但其实不是最重要的一件事情。因为它有太多的SQL语法解析，这个不是一个合适的子框架来让咱们研究。

3）为何不选Spark R？

Spark R如今很不成熟，并且支持功能有限，这个也从咱们的候选列表中删除掉。

4）为何不选Spark GraphX（图计算）？

若是你们关注了Spark的演进或发展的话，Spark最近发布的几个版本，Spark图计算基本没有改进。若是按照这个趋势的话，Spark官方机构彷佛 在透露一个信号，图计算已经发展到尽头了。因此说，咱们若是要研究的话，确定不会去作一个看上去发展到尽头的东西。另外，至于图计算而言，它有不少数学级 别的算法，而咱们是要把Spark作到极致，这样的话，数学这件事情很重要，但对咱们来讲却不是最重要的。

5）为何不选Spark MLlib（机器学习）？

Spark机器学习在封装了Vector（向量）和Metrics基础之上，加上Spark的RDD，构建了它的众多的库。这个也因为涉及到了太多的数学的知识，因此咱们选机器学习其实也不是一个太好的选择。

综上所述，咱们筛选之下，Spark Streaming是咱们惟一的选择。

我 们回顾过去，2015年是Spark最火的一年，最火的国家主要是美国。其实，2015年也是流式处理最火的一年。从从业人员的待赶上看，不论2015年 仍是2016年，在搞大数据开发的公司中，以Spark岗位招聘的待遇必定是最高的。2016上半年，据StackOverflow开展的一项调查结果显 示，在大数据领域，Spark从业人员的待遇是最高的。在调查中，50%以上的人认为，Spark中最吸引人的是Spark Streaming。总之，你们考虑用Spark，主要是由于Spark Streaming。

Spark Streaming到底有什么魔力？

1）它是流式计算

这是一个流处理的时代，一切数据若是不是流式的处理或者跟流式的处理不相关的话，都是无效的数据。这句话会不断地被社会的发展所证明。

2）流式处理才是真正的咱们对大数据的初步印象

一方面，数据流进来，当即给咱们一个反馈，这不是批处理或者数据挖掘能作到的。另外一方面，Spark很是强大的地方在于它的流式处理能够在线的利用机器学习、图计算、Spark SQL或者Spark R的成果，这得益于Spark多元化、一体化的基础架构设计。也就是说，在Spark技术堆栈中，Spark Streaming能够调用任何的API接口，不须要作任何的设置。这是Spark无可匹敌之处，也是Spark Streaming必将一统天下的根源。这个时代的流处理单打独斗已经不行了，Spark Streaming必然会跟多个Spark子框架联合起来，称霸大数据领域。

3）流式处理“魅力和复杂”的双重体

若是你精通SparkStreaming，你就知道Spark Streaming以及它背后的兄弟框架，展现了Spark和大数据的无穷魅力。不过，在Spark的全部程序中，确定是基于SparkStreaming的应用程序最容易出问题。为何？由于数据不断流进来，它要动态控制数据的流入，做业的切分还有数据的处理。这些都会带来极大的复杂性。

4）与其余Spark子框架的巨大区别

若是你仔细观察，你会发现，Spark Streaming很像是基于Spark Core之上的一个应用程序。不像其余子框架，好比机器学习是把数学算法直接应用在Spark的RDD之上，Spark Streaming更像通常的应用程序那样，感知流进来的数据并进行相应的处理。

因此若是要作Spark的定制开发，Spark Streaming则提供了最好的参考，掌握了Spark Streaming也就容易开发任意其余的程序。固然想掌握SparkStreaming，但不去精通Spark Core的话，那是不可能的。Spark Core加Spark Streaming更是双剑合璧，威力无穷。咱们选择SparkStreaming来入手，等因而找到了关键点。若是对照风水学的说法，对于Spark，咱们算是已经幸运地找到了龙脉。若是要寻龙点穴，那么Spark Streaming就是龙穴之所在。找到了穴位，咱们就能一日千里。

 

2 Spark Streaming另类在线实验

咱们在研究Spark Streaming的过程当中，会有困惑的事情：如何清晰的看到数据的流入、被处理的过程？

使用一个小技巧，经过调节放大Batch Interval的方式，来下降批处理次数，以方便看清楚各个环节。

咱们从已写过的广告点击的在线黑名单过滤的Spark Streaming应用程序入手。

 

 

import org.apache.spark.SparkConf

import org.apache.spark.streaming.{Seconds, StreamingContext} 

object OnlineBlackListFilter { def main(args: Array[String]) {  /** * 第1步：建立Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息。 * 例如说经过setMaster来设置程序要连接的Spark集群的Master的URL,若是设置 * 为local，则表明Spark程序在本地运行，特别适合于机器配置条件很是差（例如 * 只有1G的内存）的初学者 */val conf = new SparkConf() //建立SparkConf对象 conf.setAppName("OnlineBlackListFilter") //设置应用程序的名称，在程序运行的监控界面能够看到名称 conf.setMaster("spark://Master:7077") //此时，程序在Spark集群

val ssc = new StreamingContext(conf,Seconds(300))  /** * 黑名单数据准备，实际上黑名单通常都是动态的，例如在Redis或者数据库中，黑名单的生成每每有复杂的业务 * 逻辑，具体状况算法不一样，可是在Spark Streaming进行处理的时候每次都能工访问完整的信息 */

val blackList = Array(("hadoop",true),("mahout",true)) val blackListRDD = ssc.sparkContext.parallelize(blackList,8) //监听主机Master上的9999端口，接收数据val adsClickStream = ssc.socketTextStream("Master" ,9999)  /** * 此处模拟的广告点击的每条数据的格式为：time、name * 此处map操做的结果是name、（time，name）的格式 */

val adsClientStreamFormated = adsClickStream.map(ads=>(ads.split(" ")(1),ads)) adsClientStreamFormated.transform(userClickRDD => { //经过leftOuterJoin操做既保留了左侧用户广告点击内容的RDD的全部内容，又得到了相应点击内容是否在黑名单中val joinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)  /** * 进行filter过滤的时候，其输入元素是一个Tuple：（name,((time,name), boolean)） * 其中第一个元素是黑名单的名称，第二元素的第二个元素是进行leftOuterJoin的时候是否存在在值 * 若是存在的话，表面当前广告点击是黑名单，须要过滤掉，不然的话则是有效点击内容； */val validClicked = joinedBlackListRDD.filter(joinedItem=>{ if(joinedItem._2._2.getOrElse(false)){ false }else{ true } }) validClicked.map(validClick => {validClick._2._1}) }).print()  /** * 计算后的有效数据通常都会写入Kafka中，下游的计费系统会从kafka中pull到有效数据进行计费 */ ssc.start() ssc.awaitTermination() } }

　　

 

把程序的Batch Interval设置从30秒改为300秒：

 

val ssc = new StreamingContext(conf, Seconds(300))

从新生成一下jar包 。

 

Spark集群有5台机器：Master、Worker一、Worker二、Worker三、Worker4。

启动HDFS集群：start-dfs.sh
启动Spark集群：start-all.sh
启动Spark的History Server：start-history-server.sh

打开数据发送的端口：nc -lk 9999。

用spark-submit运行前面生成的jar包。

/usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --class com.dt.spark.sparkstreaming.OnlineBlackListFilter --master spark://Master:7077 /root/Documents/SparkApps/OnlineBlackListFilter.jar

在数据发送端口输入若干数据，好比：

 

1375864674543 Tom

1375864674553 Spy

1375864674571 Andy

1375864688436 Cheater

1375864784240 Kelvin

1375864853892 Steven

1375864979347 John

 

 

打开浏览器，看History Server的日志信息：

点击最新的应用，看咱们目前运行的应用程序中有些什么Job：

总共居然有5个Job。这彻底不是咱们此前作Spark SQL之类的应用程序时看到的样子。

咱们接下来看一看这些Job的内容，主要揭示一些现象，不会作彻底深刻的剖析，只是为了先让你们进行一些思考。

 

Job 0：此Job不体现咱们的业务逻辑代码。这个Job是出于对后面计算的负载均衡的考虑。

Job 0包含有Stage 0、Stage 1。随便看一个Stage，好比Stage 1。看看其中的Aggregated Metrics by Executor部分：

发现此Stage在全部Executor上都存在。

 

Job 1：运行时间比较长，耗时1.5分钟。

点击Stage 2的连接，进去看看Aggregated Metrics By Executor部分：

能够知道，Stage 2只在Worker4上的一个Executor执行，并且执行了1.5分钟。

是否会以为奇怪：从业务处理的角度看，咱们发送的那么一点数据，没有必要去启动一个运行1.5分钟的任务吧。那这个任务是作什么呢？

从DAG Visualization部分，就知道此Job实际就是启动了一个接收数据的Receiver：

原来Receiver是经过一个Job来启动的。那确定有一个Action来触发它。

看看Tasks部分：

只有一个Worker运行此Job。是用于接收数据。

Locality Level是PROCESS_LOCAL，原来是内存节点。因此，默认状况下，数据接收不会使用磁盘，而是直接使用内存中的数据。

看来，Spark Streaming应用程序启动后，本身会启动一些Job。默认启动了一个Job来接收数据，为后续处理作准备。

重要启示： 一个Spark应用程序中能够启动不少Job，而这些不一样的Job之间能够相互配合。这一认识为咱们写复杂Spark程序奠基了良好的基础。

 

Job 2：看Details能够发现有咱们程序的主要业务逻辑，体如今Stag 三、Stag四、Stag 5中。

咱们看Stag三、Stage4的详情，能够知道这2个Stage都是用4个Executor执行的。全部数据处理是在4台机器上进行的。

Stag 5只在Worker4上。这是由于这个Stage有Shuffle操做。


Job3：有Stage 六、Stage 七、Stage 8。其中Stage 六、Stage 7被跳过。

看看Stage 8的Aggregated Metrics by Executor部分。能够看到，数据处理是在4台机器上进行的：

Job4： 也体现了咱们应用程序中的业务逻辑 。有Stage 九、Stage 十、Stage 11。其中Stage 九、Stage 10被跳过。

看看Stage 11的详情。能够看到，数据处理是在Worker2以外的其它3台机器上进行的：

综合以上的现象能够知道，Spark Streaming的一个应用中，运行了这么多Job，远不是咱们从网络博客或者书籍上看的那么简单。

咱们有必要经过这些现象，反过来回溯去寻根问源。不过此次暂不作深刻分析。

咱们的神奇之旅才刚刚开始。

3 瞬间理解Spark Streaming本质

咱们先看一张图：

 

以上的连续4个图，分别对应如下4个段落的描述：

Spark Streaming接收Kafka、Flume、HDFS和Kinesis等各类来源的实时输入数据，进行处理后，处理结果保存在HDFS、Databases等各类地方。

Spark Streaming接收这些实时输入数据流，会将它们按批次划分，而后交给Spark引擎处理，生成按照批次划分的结果流。

Spark Streaming提供了表示连续数据流的、高度抽象的被称为离散流的DStream。DStream本质上表示RDD的序列。任何对DStream的操做都会转变为对底层RDD的操做。

Spark Streaming使用数据源产生的数据流建立DStream，也能够在已有的DStream上使用一些操做来建立新的DStream。

 

在咱们前面的实验中，每300秒会接收一批数据，基于这批数据会生成RDD，进而触发Job，执行处理。

 

DStream是一个没有边界的集合，没有大小的限制。

DStream表明了时空的概念。随着时间的推移，里面不断产生RDD。

锁定到时间片后，就是空间的操做，也就是对本时间片的对应批次的数据的处理。

 

下面用实例来说解数据处理过程。

从Spark Streaming程序转换为Spark执行的做业的过程当中，使用了DStreamGraph。

Spark Streaming程序中通常会有若干个对DStream的操做。DStreamGraph就是由这些操做的依赖关系构成。

从程序到DStreamGraph的转换，如如下图例所示：

本例中，从每一个foreach开始，都会进行回溯。从后往前回溯这些操做之间的依赖关系，也就造成了DStreamGraph。

执行从DStream到RDD的转换，也就造成了RDD Graph，以下图所示：

空间维度肯定以后，随着时间不断推动，会不断实例化RDD Graph，而后触发Job去执行处理。

如今再去读官方的Spark Streaming的文档，就好理解多了。

看来咱们的学习，将从Spark Streaming的现象开始，深刻到Spark Core和Spark Streaming的本质。

 

备注： 本博客内容来源于Spark发行版本定制课程