通过案例对SparkStreaming 透彻理解三板斧之一：解密SparkStreaming另类实验

从spark框架的角度来说，spark包含了如下图所示的功能:

从最近spark发布的几个版本来看，GraphX一直没有大的变化，MLlib涉及大量的数学，spark sql则是涉及了大量的sql的优化，所以如果要从某一个子框架来彻底研究spark 那么，spark streaming子框架是最好的切入口。因为本身它就很像是spark core 的一个应用程序。2015年是流式处理的一年，一切数据不是流式处理的话，那么它将失去它应有的价值。

Sparkstreaming可以调用，sparksql ,MLlib,Graphx,这些都源于它的一体化，多元化的特性。

如何去研究sparkstreaming ,将Batch interval放大，相当于看到了Streaming的慢放版本，可以更清楚它的各个环节工作流程以及状态。

packagecom.dt.spark.sparksteaming

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

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

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

      valadsClickStream = ssc.socketTextStream("Master", 9999)

      /**
       *此处模拟的广告点击的每条数据的格式为：time、name
       *此处map操作的结果是name、（time，name）的格式
       */
      valadsClickStreamFormatted = adsClickStream.map { ads => (ads.split("")(1), ads) }
      adsClickStreamFormatted.transform(userClickRDD=> {
        //通过leftOuterJoin操作既保留了左侧用户广告点击内容的RDD的所有内容，又获得了相应点击内容是否在黑名单中
        valjoinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)

        /**
         *进行filter过滤的时候，其输入元素是一个Tuple：（name,((time,name), boolean)）
         *其中第一个元素是黑名单的名称，第二元素的第二个元素是进行leftOuterJoin的时候是否存在在值
         *如果存在的话，表面当前广告点击是黑名单，需要过滤掉，否则的话则是有效点击内容；
         */
        valvalidClicked = 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()

    }
}

打包运行于spark 集群之中，具体方法如下：

1首先启动hdfs和spark集群，并且开启historyServer以便于查看执行完成的job

2将上述代码打成jar包

3.运行该命令： /usr/local/spark-1.6.1-bin-hadoop2.4/bin/spark-submit --classcom.pzw.spark.OnlineBlackListFilter --master spark://Master:7077/root/Documents/SparkApps/WordCount.jar

Note:需要启动nc -lk 9999

从spark ui 的角度分析：

查看JOB0 的UI：

从此处可以得出，spark在启动的时候会自动创建一些job.

这个job的作用是为了最大化第利用集群资源。

继续看jobid1

此处任务接受者以任务的方式运行在worker中的executor中，并且只有一个task运行。从tasks提示来看，localitylevel为process_local ,我们之前提到过，sparkstreaming 保存数据的方式是memory_and_disk_ser_2的方式，所以当内存能够放下我们接受到的数据的时候，是不会保存到磁盘中的。这里的数据要经过网络传输，每次数据收集的时候会将数据分片，并
且将数据分发到各个计算节点上。

Jobid2