spark streaming (二)

首页 > 代码库 > spark streaming (二)

2024-09-11 06:28:53 216人阅读

一、基础核心概念

1、StreamingContext详解（一）

有两种创建StreamingContext的方式：
val conf = new SparkConf().setAppName(appName).setMaster(master);
val ssc = new StreamingContext(conf, Seconds(1));

StreamingContext，还可以使用已有的SparkContext来创建
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(1));

appName，是用来在Spark UI上显示的应用名称。 master，是一个Spark、 Mesos或者Yarn集群的URL，或者是local[*]。

2、StreamingContext详解（二）

一个StreamingContext定义之后，必须做以下几件事情：
1、通过创建输入DStream来创建输入数据源。
2、通过对DStream定义transformation和output算子操作，来定义实时计算逻辑。
3 、调用StreamingContext的start()方法，来开始实时处理数据。
4、调用StreamingContext的awaitTermination()方法，来等待应用程序的终止。可以使用CTRL+C手动停止，或者就是让它持续不断的运行进行计算。
5、也可以通过调用StreamingContext的stop()方法，来停止应用程序。

需要注意的要点：
1、只要一个StreamingContext启动之后，就不能再往其中添加任何计算逻辑了。比如执行start()方法之后，还给某个DStream执行一个算子。
2、一个StreamingContext停止之后，是肯定不能够重启的。调用stop()之后，不能再调用start()
3、一个JVM同时只能有一个StreamingContext启动。在你的应用程序中，不能创建两个StreamingContext。
4、调用stop()方法时，会同时停止内部的SparkContext，如果不希望如此，还希望后面继续使用SparkContext创建其他类型的Context，比如SQLContext，那么就用stop(false)。
5、一个SparkContext可以创建多个StreamingContext，只要上一个先用stop(false)停止，再创建下一个即可。

3、输入DStream和Receiver详解（一）

输入DStream代表了来自数据源的输入数据流。在之前的wordcount例子中， lines就是一个输入 DStream（ JavaReceiverInputDStream），代表了从netcat（ nc）服务接收到的数据流。除了文件数据流之外，所有的输入DStream都会绑定一个Receiver对象，

该对象是一个关键的组件，用来从数据源接收数据，并将其存储在Spark的内存中，以供后续处理。
Spark Streaming提供了三种内置的数据源支持；
1、基础数据源： StreamingContext API中直接提供了对这些数据源的支持，比如文件、 socket、 Akka Actor等。
2、高级数据源：诸如Kafka、 Flume、 Kinesis、 Twitter等数据源，通过第三方工具类提供支持。这些数据源的使用，需要引用其依赖。
3、自定义数据源：我们可以自己定义数据源，来决定如何接受和存储数据。

4、输入DStream和Receiver详解（二）

如果你想要在实时计算应用中并行接收多条数据流，可以创建多个输入DStream。这样就会创建多个 Receiver，从而并行地接收多个数据流。但是要注意的是，一个Spark Streaming Application的 Executor，是一个长时间运行的任务，因此，

它会独占分配给Spark Streaming Application的cpu core。从而只要Spark Streaming运行起来以后，这个节点上的cpu core，就没法给其他应用使用了。

使用本地模式，运行程序时，绝对不能用local或者local[1]，因为那样的话，只会给执行输入DStream的 executor分配一个线程。而Spark Streaming底层的原理是，至少要有两条线程，一条线程用来分配给 Receiver接收数据，一条线程用来处理接收到的数据。

因此必须使用local[n]， n>=2的模式。（n不能大于当前节点的CPU核数）
如果不设置Master，也就是直接将Spark Streaming应用提交到集群上运行，那么首先，必须要求集群节点上，有>1个cpu core，其次，给Spark Streaming的每个executor分配的core，必须>1，这样，才能保证分配到executor上运行的输入DStream，

两条线程并行，一条运行Receiver，接收数据；一条处理数据。否则的话，只会接收数据，不会处理数据。

总结：Receiver接收器

Receiver接收器，可以接收外部数据源中的数据，并将其保存到
内存中，以供后续使用。
在Spark中支持三大类型的数据源：
1、基础数据源：比如文件、Socket、Akka中的数据。
2、高级数据源，比如Flume、Kafka、推特中的数据。
3、自定义数据源。
补充：在Spark Streaming中，可以通过两种方式操作Kafka的数据。
一种是通过Receiver的方式，另一种Direct直接读取的方式。

5、输入DStream之基础数据源
（1）Socket：之前的wordcount例子，

object WordCoundStreaming {
  def main(args: Array[String]): Unit = {
    /**
      * 处理Ｓpakr Streaming程序至少需要２个线程，其中，
      * 一个线程负责接收输入的数据
      * 另一个线程负责处理接收的数据
      * local[N]　N>=2
      */
    val conf=new SparkConf().setAppName("WordCoundStreaming")
              .setMaster("local[2]")
    /**
      * SparkContext是用户与Spark集群交互的唯一接口，所以SparkContext是必需的。
      * 在创建StreamingContext的过程中，Spark会在源码中自动创建一个SparkContext对象。
      * 注意第二个参数Seconds(＊)，表示实时流数据中每批数据的时间间隔，
      * 也就是说，在DStream离散流中的每个RDD包含相应时间间隔的数据。
      */
    val ssc=new StreamingContext(conf,Seconds(5))
    /**
      * 通过socketTextStream()获取nc服务器中的数据
      * 需要指明获取nc服务器的节点名称和端口，这是的端口要和运行nc服务器的端口一致。
      * 此时产生的lines不是RDD，而是一个DStream离散流。
      */
    val lines=ssc.socketTextStream("tgmaster",9999)
    /**
      * 对离散流lines进行flatMap转换操作，实际上是对lines离散流中的每个RDD都进行
      * flatMap操作，从而产生了新的RDD，多个新的RDD构成了新的离散流words。
      */
    val words=lines.flatMap(_.split(" "))
    val pairs=words.map((_,1))
    val result=pairs.reduceByKey(_+_)

    //在控制台输出内容
    result.print()
    //启动StreamingContext
    ssc.start()
    //等待程序停止
    ssc.awaitTermination()
  }
}

　　（2）基于HDFS文件的实时计算，其实就是，监控一个HDFS目录，只要其中有新文件出现，就实时处理。相当于处理实时的文件流。
streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory)
streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
Spark Streaming会监视指定的HDFS目录，并且处理出现在目录中的文件。
要注意的是，所有放入HDFS目录中的文件，都必须有相同的格式；必须使用移动或者重命名的方式，将文件移入目录；一旦处理之后，文件的内容即使改变，也不会再处理了；基于HDFS文件的数据源是没有Receiver的，因此不会占用一个cpu core。

 def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("word").setMaster("local[2]")
    val ssc=new StreamingContext(conf,Seconds(5))
    val lines=ssc.textFileStream("hdfs://liuwei1:9000/homework")
    val result=lines.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    result.print()
    ssc.start()
    ssc.awaitTermination()
  }

二、DStream的transformation

技术分享

总结：DStream中Transformation类型的算子有以下三个最重要
1)updateStateByKey
可以为每个Key保留一份状态，并更新状态的值。
案例：在全局范围内统计每个单词出现的次数。
2)transform
可以执行RDD到RDD的操作，相当于对DStream API的一个补充。
3)window
滑动窗口操作，需要指明两个参数，一个是窗口的长度，另一个是
窗口滑动的时间间隔。

三、updateStateBykey （全局范围之内处理数据，而不是一批一批的）
updateStateByKey
updateStateByKey操作，可以让我们为每个key维护一份state，并持续不断的更新该state。
1、首先，要定义一个state，可以是任意的数据类型；
2、其次，要定义state更新函数——指定一个函数如何使用之前的state和新值来更新state。
对于每个batch， Spark都会为每个之前已经存在的key去应用一次state更新函数，无论这个key在batch中是否有新的数据。如果state更新函数返回none，那么key
对应的state就会被删除。当然，对于每个新出现的key，也会执行state更新函数。
注意， updateStateByKey操作，要求必须开启Checkpoint机制。
案例：基于缓存的实时wordcount程序（在实际业务场景中，这个是非常有用的）

object updateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("updateStateByKey")
            .setMaster("local[2]")
    val ssc=new StreamingContext(conf,Seconds(5))
    //updateStateByKey操作，要求必须开启Checkpoint机制。
    ssc.checkpoint("hdfs://tgmaster:9000/in/ch")

    val lines=ssc.socketTextStream("tgmaster",9999)
    val pairs=lines.flatMap(_.split(" "))
              .map((_,1))
    /**
      * values:多个新值的集合
      * state：是一个Option类型的状态，
      * 可以通过state.getOrElse(0)为newValue设置初始值
      */
    val result=pairs.updateStateByKey((values:Seq[Int],state:Option[Int])=>{
      //创建newValue并设初始值０
      var newValue=http://www.mamicode.com/state.getOrElse(0)>

四、transform

transform操作，应用在DStream上时，可以用于执行任意的RDD到RDD的转换操作。它可以用于实现， DStream API中所没有提供的操作。比如说， DStream API中，并没有提供将一个DStream中的每个batch，与一个特定的RDD进行join的操作。

但是我们自己就可以使用transform操作来实现该功能。
DStream.join()，只能join其他DStream。在DStream每个batch的RDD计算出来之后，会去跟其他DStream的RDD进行join。
案例：广告计费日志实时黑名单过滤

/**
  * Created by tg on 3/29/17.
  * 实时黑名单过滤
  */
object transformDemo {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("transformDemo")
                .setMaster("local[2]")
    val ssc=new StreamingContext(conf,Seconds(5))

    //模拟数据，创建黑名单RDD,(String,Boolean) (name,Boolean)
    val blackRDD=ssc.sparkContext.parallelize(Array(("tom",true)))
    //从nc服务器中获取数据，nc服务器中的数据格式：time name,比如: 1101 jack
    val linesDStream=ssc.socketTextStream("tgmaster",9999)
    val mapDStream=linesDStream.map(line=>{
      val log=line.split(" ")
      (log(1),line) //(name,time name)
    })
    /**
      * tansform()算子可以执行RDD到RDD的转换操作
      */
    mapDStream.transform(adsRDD=>{
      //让adsRDD与blackRDD进行leftOuterJoin左外连接操作，操作之后的数据包含所有的用户
     // (String,(String,Boolean))
      val joinRDD=adsRDD.leftOuterJoin(blackRDD)
      val filterRDD=joinRDD.filter(x=>{
        /**
          * 当if条件成立时，意味着是黑名单人员，返回false将其删除。
          * 当if条件不成立时，意味着不是黑名单人员，返回true将其保留。
          */
        if(x._2._2.getOrElse(false)) false else true
      })
      filterRDD.map(x=>{
          x._2._1
      })
    }).print()

    ssc.start()
    ssc.awaitTermination()
  }
}

五、window滑动窗口

Spark Streaming提供了滑动窗口操作的支持，从而让我们可以对一个滑动窗口内的数据执行计算操作。
每次掉落在窗口内的RDD的数据，会被聚合起来执行计算操作，然后生成的RDD，会作为window DStream的一个RDD。比如下图中，就是对每三秒钟的数据执行一次滑动窗口计算，这3秒内的3个 RDD会被聚合起来进行处理，然后过了两秒钟，又会对最近三秒内的数据执行滑动窗口计算。所以每个滑动窗口操作，都必须指定两个参数，窗口长度以及滑动间隔，而且这两个参数值都必须是 batch间隔的整数倍。（ Spark Streaming对滑动窗口的支持，是比Storm更加完善和强大的）
技术分享

window滑动窗口操作

案例：
热点搜索词滑动统计，每隔5秒钟，统计最近20秒钟的搜索词的搜索频次，并打印出排名最靠前的3个搜索词以及出现次数

package SparkCore.day1

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by tg on 3/29/17.
  * 热点搜索词滑动统计，每隔5秒钟，统计最近20秒钟的搜索词的搜索频次，
  * 并打印出排名最靠前的3个搜索词以及出现次数。
  */
object reduceByKeyAndWindowDemo {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("reduceByKeyAndWindowDemo")
              .setMaster("local[2]")
    val ssc=new StreamingContext(conf,Seconds(1))
    //从nc服务器中获取数据，数据格式：name keyword,比如：张三 Spark
    val linesDStream=ssc.socketTextStream("tgmaster",9999)
    //取出每行的搜索关键词
    val wordsDStream=linesDStream.map(x=>x.split(" ")(1))
    val pairsDStream=wordsDStream.map(x=>(x,1))
    /**
      * 通过滑动窗口进行统计，每隔5秒钟，统计最近20秒钟的搜索词出现次数。
      * reduceByKeyAndWindow()算子中，
      * 第一部分v1+v2用于计算搜索词出现次数
      * 第二部分Seconds(20)用于设置窗口的长度
      * 第三部分Seconds(5)用于设置窗口的时间间隔
      */
    val resultDStream=pairsDStream.reduceByKeyAndWindow((v1:Int,v2:Int)=>
    v1+v2,Seconds(20),Seconds(5))

    resultDStream.transform(itemRDD=>{
      val result=itemRDD.map(x=>(x._2,x._1))
        .sortByKey(false).take(3)
        .map(x=>(x._2,x._1))

      val resultRDD=ssc.sparkContext.parallelize(result)
      resultRDD
    }).print()

    ssc.start()
    ssc.awaitTermination()
  }
}

spark streaming (二)

声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉：投诉/举报工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

联系
我们

首页 > 代码库 > spark streaming (二)

spark streaming (二)

看完仍有疑问？有类似问题直接问程序猿