第1课：解密SparkStreaming另类实验及SparkStreaming本质解析

本期内容

1 sparkStreaming另类在线实验

2瞬间理解sparkStreaming本质

sparkStreaming本身是随着流进来数据，按照时间为单位生成job,触发job在 clustr执行的流式处理引擎。（解偶合）sparkStreaming本质是加上了时间维度的批处理。

在流处理时代，SparkStreaming有着强大吸引力，而且发展前景广阔，加之Spark的生态系统，Streaming可以方便调用其他的诸如SQL，MLlib等强大框架，它必将一统天下。这里选择Spark Streaming作为版本定制的切入点也是大势所趋。
小技巧：将Batch interval放大，相当于看到了Streaming的慢放版本，可以更清楚它的各个环节，这里以黑名单过滤程序为例，进行试验

案例源码

package com.dt.spark.sparksteaming
 
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
/**
 *
使用Scala开发集群运行的Spark 在线黑名单过滤程序
 * @author DT大数据梦工厂
 *
新浪微博：http://weibo.com/ilovepains/
 *
 *
背景描述：在广告点击计费系统中，我们在线过滤掉黑名单的点击，进而保护广告商的利益，只进行有效的广告点击计费
 * 
或者在防刷评分（或者流量）系统，过滤掉无效的投票或者评分或者流量；
 *
实现技术：使用transform Api直接基于RDD编程，进行join操作
 *
 */
object OnlineBlackListFilter {
    def main(args: Array[String]){
     
/**
      
* 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息
      
*/
     
val conf = new SparkConf() //创建SparkConf对象
     
conf.setAppName("OnlineBlackListFilter") //设置应用程序的名称，在程序运行的监控界面可以看到名称
     
conf.setMaster("spark://Master:7077") //此时，程序在Spark集群
       

     
val ssc = new StreamingContext(conf, Seconds(30))//这里可以将Batch interval调整到更大，例如300秒，以便更好的了解Streaming内幕
      

     
/**
      
* 黑名单数据准备，实际上黑名单一般都是动态的，例如在Redis或者数据库中，黑名单的生成往往有复杂的业务
      
* 逻辑，具体情况算法不同，但是在Spark Streaming进行处理的时候每次都能工访问完整的信息
      
*/
     
val blackList = Array(("hadoop", true),("mahout", true))
     
val blackListRDD = ssc.sparkContext.parallelize(blackList, 8)
      

     
val adsClickStream = ssc.socketTextStream("Master", 9999)
      

     
/**
      
* 此处模拟的广告点击的每条数据的格式为：time、name
      
* 此处map操作的结果是name、（time，name）的格式
      
*/
     
val adsClickStreamFormatted = adsClickStream.map { ads => (ads.split(" ")(1), ads) }
     
adsClickStreamFormatted.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()
     

    }
}
 

运行分析

启动hdfs和spark集群并启动historyServer，将上述代码打成jar包放入/project/bin/目录下，为方便这里还是命名为robinspark-1.0-SNAPSHOT.jar。编辑脚本文件start.sh 如下：
$SPARK_HOME/bin/spark-submit \

        --class com.dt.spark.sparkstreaming.OnlineBlackListFilter \

        --master spark://Master:7077 \

        ./robinspark-1.0-SNAPSHOT.jar

        --f
 
运行脚本文件
root@Master:/project/bin# sh -x start.sh
注意：
这里需要启动nc -lk 9999否则会报错
root@Master:/# nc -lk 9999
22222 Spark
3333 flink
4444 mahout
1111 spark
444444 hadoop
5555 kafka
6666 hdfs
8888 java

运行结果：

执行完毕去SparkUI查看job

点击进入DAG图

 

从DAG图中可以看到这里并不是应用程序的逻辑，这说明Spark Streaming本身更像是一个应用程序，它在启动的时候会自动启动一些作业，执行好几个job。这时进入details，会发现有一个Receiver在接受数据，而且有一个task运行了7.7min，而打开historyServer会发现整个应用程序也就执行了9.4min。这个7.7min的task就是Receiver在不断循环的接受数据。从这里可以看出，Spark
Streaming启动Receiver是通过job启动的，Receiver接受数据和普通job没有区别。我们从一台机器上接受数据，可以在多个机器上执行，最大化的利用资源。在整个过程中虽然产生了很多job，但真正只有一个作业在执行。
Spark Streaming本身是随着流进来的数据，按照时间为单位生成job，触发job在cluster上执行的流式处理引擎。从实质上讲，它是加上了时间维度的批处理。每隔一段时间，就会有一批数据流入，通过DStream模版不断的产生RDD，触发job并处理。
 

面试题：

你写的sparkStreaming运行多少个job，这些job是怎么配合的？

（Duration）这个job就是指reciver, reciver是在excuter上运行的，也就是说sparkStreaming框架在启动reciver的时候是通过一个job启动的，肯定有一个action触发。

可以启动多个reciver吗？可以启动多个reciver

接收数据在一个work中发生的，但是在多台数据上进行处理。普通的reciver和job没有区别

DStream是指随着时间的推进不断的产生RDD

在一个spark application中可以启动很多job,不同的job之间可以相互配合。

 

 

博主：罗白莲

资料来源于：王家林（Spark版本定制班课程）

新浪微博：http://www.weibo.com/ilovepains