第1课：SparkStreaming 三板斧之一：解密SparkStreaming另类实验及SparkStreaming本质解析

本期内容：

SparkStreaming在线另类实验

瞬间理解SparkStreaming的本质

SparkStreaming 是Spark Core上的一个子框架，如果我们能够完全精通了一个子框架，我们就能够更好的驾驭Spark。SparkStreaming和Spark SQL是目前最流行的框架，从研究角度而言，Spark SQL有太多涉及到SQL优化的问题，不太适应用来深入研究。而SparkStreaming和其他的框架不同，它更像是SparkCore的一个应用程序。如果我们能深入的了解SparkStreaming，那我们就可以写出非常复杂的应用程序。
SparkStreaming的优势是可以结合SparkSQL、图计算、机器学习，功能更加强大。这个时代，单纯的流计算已经无法满足客户的需求啦。在Spark中SparkStreaming也是最容易出现问题的，因为它是不断的运行，内部比较复杂。

本次实验基于如下博客中的程序代码
IMF课程的第94课：SparkStreaming 实现广告计费系统中在线黑名单过滤实战http://lqding.blog.51cto.com/9123978/1769290

为了更好的查看Job的运行情况，我们启动history-server

root@spark-master:/usr/local/spark/spark-1.6.0-bin-hadoop2.6/sbin# ./start-history-server.sh

history-server启动失败，查看日志报如下信息：

Exception in thread "main" java.lang.reflect.InvocationTargetException
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at org.apache.spark.deploy.history.HistoryServer$.main(HistoryServer.scala:235)
at org.apache.spark.deploy.history.HistoryServer.main(HistoryServer.scala)
Caused by: java.lang.IllegalArgumentException: Log directory specified does not exist: file:/tmp/spark-events. Did you configure the correct one through spark.history.fs.logDirectory?

根据报错信息，大致可以看出是Log目录不存在。创建目录

root@spark-master:/tmp# hdfs dfs -mkdir /historyServerForSpark/

配置spark-env.sh，添加一个环境变量，让history server的logDirectory指向上面建立的目录

export SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://spark-master:8020/historyServerForSpark"

配置spark-defaults.conf，添加如下配置项：

#是否记录作业产生的事件或者运行状态(job，stage等使用内存等信息)
spark.eventLog.enabled           true
#如果记录作业产生的事件或者运行状态，则将事件写入什么位置
spark.eventLog.dir             hdfs://spark-master:8020/historyServerForSpark
#http history的监听端口号，通过http://hadoop.master:18080访问
spark.history.ui.port            18080
#Spark history日志位置
park.history.fs.logDirectory=hdfs://spark-master:8020/historyServerForSpark

再次启动history-server，问题解决。
WEB界面如下：



为了可以更清晰的看清楚Streaming运行的各个环节，我们可以通过将batchInterval的值设置的更大。例如5分钟。

将程序上传至spark集群

运行Spark程序

root@spark-master:~# /usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --class com.dt.spark.streaming.OnlineBlackListFilter --master spark://spark-master:7077 ./spark.jar

打开netcat，并发送一些数据，内容如下：

root@spark-master:~# nc -lk 9999
134343 Hadoop
343434 spark
3432777 Java
0983743 Hbase
893434 Mathou

程序输入结果为：

16/05/01 14:00:01 INFO scheduler.DAGScheduler: Job 3 finished: print at OnlineBlackListFilter.scala:63, took 0.098316 s
-------------------------------------------
Time: 1462082400000 ms
-------------------------------------------
3432777 Java
343434 spark
0983743 Hbase

计算结果出来后，将SparkStreaming程序终止掉。

接下来，我们查看web ui中的内容，来解析SparkStreaming的运行过程。



红色部分为我们刚刚运行的程序的日志(第一次运行时，在completed application这个地方看不到日志，在Show incomplete applications 这个地方显示了日志，可是此时程序已经退出了。)

我们点击进去，查看详细信息：


我们可以看到，这个程序在运行期间，启动了4个Job。
先看看job id 为0 的详细信息


这个job，很明显是我们定义的blackListRDD数据的生成。对应的代码为

val blackList = Array(("Hadoop", true), ("Mathou", true))
//把Array变成RDD
val blackListRDD = ssc.sparkContext.parallelize(blackList)

并且它做了reduceBykey的操作(代码中并没有此步操作，SparkStreaming框架自行生成的)。
这里有两个Stage，Stage 0和Stage 1 。

接下来我们看看Job 1的详细信息


此处也是一个makeRDD，这个RDD是receiver不断的接收数据流中的数据，在时间间隔达到batchInterval后，将所有数据变成一个RDD。并且它的耗时也是最长的，59s 。

特别说明：此处可以看出，receiver也是一个独立的job。由此我们可以得出一个结论：我们在应用程序中，可以启动多个job，并且不用的job之间可以相互配合，这就为我们编写复杂的应用程序打下了基础。

我们点击上面的start at OnlineBlackListFilter.scala:64查看详细信息


根据上图信息，只有一个Executor在接收数据，最最重要的是红色框中的数据本地性为PROCESS_LOCAL,由此可以知道receiver接收到数据后会保存到内存中，只要内存充足是不会写到内存中的。
即便在创建receiver时，指定的存储默认策略为MEMORY_AND_DISK_SER_2

def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String] = withNamedScope("socket text stream") {
socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)
}

我们再看看job 2的详细信息：





Job 2 将前两个job生成的RDD进行leftOuterJoin操作。
从Stage Id的编号就可以看出，它是依赖于上两个Job的。

Receiver接收数据时是在spark-master节点上，但是Job 2在处理数据时，数据已经到了spark-worker1上了（因为我的环境只有两个worker，数据并没有分散到所有worker节点，worker节点如果多一点，情况可能不一样，每个节点都会处理数据）

点击上面的Stage Id 3查看详细信息：


在一个Executor上运行，并且有5个Task 。

我们看看Job 3的详细信息：






此处的DAG图和Job2的相同，但是Stage 6和7被跳过了。详细的原因，我们后面的课程会一一讲解。

总结：我们可以看出，一个batchInterval并不是单单触发一个Job。

根据上面的描述，我们更细致的了解了DStream和RDD的关系了。DStream就是一个个batchInterval时间内的RDD组成的。只不过DStream带上了时间维度，是一个无边界的集合。




对DStream的操作会构建成DStream Graph





在每到batchInterval时间间隔后，Job被触发，DStream Graph将会被转换成RDD Graph




备注：1、DT大数据梦工厂微信公众号DT_Spark
2、IMF晚8点大数据实战YY直播频道号：68917580
3、新浪微博: http://www.weibo.com/ilovepains