1.1 Spark Streaming应用案例_Spark Streaming技术内幕及源码剖析-QQ阅读男生都市网

上QQ阅读APP看书，第一时间看更新

1.1 Spark Streaming应用案例

Spark Streaming应用程序运行的时候，往往在短时间内会产生大量日志信息，不利于研究分析。可以通过加大批处理时间间隔（batch interval）来降低批处理频率，减少日志信息量，以便看清楚各个环节。

下面从一个Spark Streaming应用程序的开发入手，观察运行过程，以增强感性认识。以下是一个广告点击的在线黑名单过滤的Spark Streaming应用程序，程序中有详细注释，以方便初次接触Spark Streaming的读者理解。

源码1-1 OnlineBlackListFilter

    package com.dt.spark.streaming


    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.StreamingContext
    import org.apache.spark.streaming.Seconds


    object  OnlineBlackListFilter {
        def main(args: Array[String]){
          /**
            * 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
            * 例如，通过setMaster来设置程序要链接的Spark集群的Master的URL，如果设置
            * 为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差（例如
            * 只有1GB内存）的初学者
            */
          // 创建SparkConf对象
          val conf = new SparkConf()
          // 设置应用程序的名称，在程序运行的监控界面可以看到名称
          conf.setAppName("OnlineBlackListFilter")
          // 此时，程序在Spark集群
          conf.setMaster("spark://Master:7077")
    val ssc = new StreamingContext(conf, Seconds(30))


    /**
      * 黑名单数据准备。实际上黑名单一般都是动态的，例如在Redis或者数据库中
      * 黑名单的生成往往有复杂的业务逻辑，具体情况算法不同
      * 但是在Spark Streaming进行处理的时候每次都能够访问完整的信息
      */
    val blackList = Array(("Spy", true), ("Cheater", 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过滤的时候，其输入元素是一个元组：（name, ((time, name), boolean)）
        * 其中，第一个元素是黑名单的名称
    * 第二元素的第二个元素boolean是进行leftOuterJoin的时候是否存在的值。
    * 如果存在，表明当前广告点击是黑名单，需要过滤掉，否则是有效点击内容
    */
  val validClicked = joinedBlackListRDD.filter(joinedItem => {
      if(joinedItem._2._2.getOrElse(false))
      {
        false
      } else {
        true
      }


  })


        validClicked.map(validClick => {validClick._2._1})
}).print


        ssc.start()
        ssc.awaitTermination()


        }
}

此程序接收Socket信息，过滤掉其中名称为Spy、Cheater的信息，并打印输出。

把程序的批处理时间间隔设置从30s改成300s：

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

然后重新生成一下jar包。

Spark集群有5台机器：Master、Worker1、Worker2、Worker3、Worker4。

启动Spark的History Server。

打开数据发送的端口：

nc -lk 9999

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

在数据发送端口输入若干数据，例如：

    1375864674543 Tom
    1375864674553 Spy
    1375864674571 Andy
    1375864688436 Cheater
    1375864784240 Kelvin
    1375864853892 Steven
    1375864979347 John

每行第一项为时刻的毫秒数，第二项是程序中要过滤的名称。

打开浏览器，看History Server的日志信息，如图1-2所示。