通过案例对SparkStreaming 透彻理解三板斧之一

378

　　王家林：DT大数据梦工厂创始人和首席专家.
　　联系邮箱18610086859@126.com
　　电话：18610086859 QQ:1740415547 微信号：18610086859
　　spark的核心是spark core,spark streaming，spark graph,spark mlib 其实是构建在spark core之上的一个应用程序，如果要构建一个强大的spark应用程序 ，spark streaming 是一个值得借鉴的参考，spark streaming涉及多个job交叉配合，里面涉及到了spark的所有的核心组件，如果对spark streaming 精通了的话，可以说就精通了整个spark,所以精通掌握spark streaming是至关重要的。
　　spark另类实现。在试验中。通过调节放大Batch Interval的方式300s，观察数据流入流出。
　　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(300))
　　/**
　　* 黑名单数据准备，实际上黑名单一般都是动态的，例如在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()
　　}
　　}
　　打开9999端口 nc -lk 9999
　　输入测试数据
　　2255554 Spark
　　455554444 Hadoop
　　55555 Flink
　　66666 Kafka
　　6666855 RockySpark
　　666638 Scala
　　66666 DT_Spark
　　在浏览器history_server中查看

　　点击最上面的任务后

　　总共竟然有5个Job。
　　job0主要用于负载均衡
　　job1的时间是1.5min， 从各个job所花的时间来看，整个应用运行的时间大部分时间是花在了receiver上
　　job2的时间

　　Spark Streaming的一个应用中，运行了这么多Job