Spark Streaming另类在线实验

本期内容：另类在线实验：将batchInterval设置得足够的大，以此来放大各个环节的细节。在实验的基础上...



感谢DT大数据梦工厂支持提供以下内容，DT大数据梦工厂专注于Spark发行版定制。

内容概括：

本期内容：另类在线实验：将batchInterval设置得足够的大，以此来放大各个环节的细节。在实验的基础上来谈Spark Streaming的本质

1 实验代码如下：在线广告黑名单过滤（Spark Streaming接收来自Kafka的数据实验）

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程序的运行时的配置信息，

* 例如说通过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(30))

/**

* 黑名单数据准备，实际上黑名单一般都是动态的，例如在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()

}

}

运行代码后，接收数据如下：



我们来看看UI界面显示信息，显示信息如下所示



这里一共有5个Job，说明SparkStreaming在启动的时候，自己会启动一些job(实际执行的是一个Job),在我们自己写的代码中没有reduceByKey等转换，UI界面里的Job中有reduceByKey操作。这是SparkStreaming框架自动启动的额。

job1,makeRDD at ReceiverTracker

图中显示作业Job运行了1.5Min，但是我们的程序中并没有执行1.5Min，这是为什么呢？因为这是启动的Receiver这个Job运行了1.5Min.这里有点小激动。



这里可以说明的是：Spark Streaming是Spark Core上的一个应用程序，一个优秀的Spark Core上的框架，一定是多个Job配合执行的。一个好的面试题目是：SparkStreaming中的多个Job是怎么配合的？（这个是非常高级的题目哦）（通过Stage进行衔接）

实质上：Spark Streaming只有一个JOB执行，但是却产生了很多的job,和一般的job貌似不一样，这也就是Spark Streaming不是看上去那么简单的原因了。

    关注 数模堂