参考链接：http://spark.apache.org/docs/latest/streaming-programming-guide.html

一、Spark Streaming 介绍

Spark Streaming是核心Spark API的扩展，可实现实时数据流的可伸缩，高吞吐量，容错流处理。数据可以从Kafka、ZeroMQ等消息队列以及TCP sockets或者目录文件从数据源获取数据，并且可以使用map，reduce，join和window等高级函数进行复杂算法的处理。最后，可以将处理后的数据推送到文件系统，数据库和实时仪表板。

• 在内部，它的工作方式为：Spark Streaming接收实时输入数据流，并将数据分成批次，然后由Spark引擎进行处理，以生成批次的最终结果流。
• 
• 对应的批数据，在Spark内核对应一个RDD实例，因此，对应流数据的DStream可以看成是一组RDDs，即RDD的一个序列。通俗点理解的话，在流数据分成一批一批后，通过一个先进先出的队列，然后 Spark Engine从该队列中依次取出一个个批数据，把批数据封装成一个RDD，然后进行处理，这是一个典型的生产者消费者模型，对应的就有生产者消费者模型的问题，即如何协调生产速率和消费速率。 

Spark Streaming Wordcount：

• import org.apache.spark.SparkConf
  
  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  
  import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
  
  /**
  
    * @author xiandongxie
  
    */
  
  object SparkStreamingWordCount {
  
    def main(args: Array[String]): Unit = {
  
      val conf: SparkConf = new SparkConf()
  
        .setMaster("local[2]")
  
        .setAppName("SparkStreamingWordCount")
  
      // 设置批次时间5S
  
      val duration: Duration = Durations.seconds(5)
  
      val context: StreamingContext = new StreamingContext(conf,duration)
  
      // 指定socket数据源
  
      val sourceDStream: ReceiverInputDStream[String] = context.socketTextStream("localhost", 6666)
  
      // 计算WordCount
  
      val resultDStream: DStream[(String, Int)] = sourceDStream.flatMap(f => f.split("\t"))
  
        .map((_, 1))
  
        .reduceByKey(_ + _)
  
      resultDStream.print()
  
      context.start()             // Start the computation
  
      context.awaitTermination()  // Wait for the computation to terminate
  
    }
  
  }

二、Spark Streaming 对比 Storm

• 处理模型以及延迟
  • 虽然两框架都提供了可扩展性(scalability)和可容错性(fault tolerance)，但是它们的处理模型从根本上说是不一样的。Storm可以实现亚秒级时延的处理，而每次只处理一条event，而Spark Streaming可以在一个短暂的时间窗口里面处理多条(batches)Event。所以说Storm可以实现亚秒级时延的处理，而Spark Streaming则有一定的时延。
• 容错和数据保证
  • 然而两者的都有容错时候的数据保证，Spark Streaming的容错（通过血缘关系）为有状态的计算提供了更好的支持。在Storm中，每条记录在系统的移动过程中都需要被标记跟踪，所以Storm只能保证每条记录最少被处理一次，但是允许从错误状态恢复时被处理多次。这就意味着可变更的状态可能被更新两次从而导致结果不正确。
  • Spark Streaming的容错：通过血缘关系，是粗粒度的，保证每个批处理记录仅仅被处理一次，即使是node节点挂掉
  • Storm：细粒度的容错，每条记录在系统的移动过程中都需要被标记跟踪，缺点：允许从错误状态恢复时被处理多次。这就意味着可变更的状态可能被更新两次从而导致结果不正确。
• 批处理框架集成
  • Spark Streaming的一个很棒的特性就是它是在Spark框架上运行的。这样你就可以使用spark的批处理代码一样来写Spark Streaming程序，或者是在Spark中交互查询比如spark-sql。这就减少了单独编写流处理程序和历史数据处理程序。
• 生产支持
  • 两者都可以在各自的集群框架中运行，但是Storm可以在Mesos上运行, 而Spark Streaming可以在YARN和Mesos上运行。

 Spark Streaming优缺点：

• 优点：
  • 吞吐量大、速度快。
  • 容错：SparkStreaming在没有额外代码和配置的情况下可以恢复丢失的工作。checkpoint。
  • 社区活跃度高。生态圈强大。因为后台是Spark
  • 数据源广泛。
• 缺点：
  • 延迟。500毫秒已经被广泛认为是最小批次大小。所以实际场景中应注意该问题，就像标题分类场景，设定的0.5s一批次，加上处理时间，分类接口会占用1s的响应时间。实时要求高的可选择使用其他框架。

三、架构与抽象

Spark Streaming使用“微批次”的架构，把流试计算当成一系列连接的小规模批处理来对待，Spark Streaming从各种输入源中读取数据，并把数据分成小组的批次，新的批次按均匀的时间间隔创建出来，在每个时间区间开始的时候，一个新的批次就创建出来，在该区间内收到的数据都会被添加到这个批次中，在时间区间结束时，批次停止增长。时间区间的大小是由批次间隔这个参数决定的，批次间隔一般设在500毫秒到几秒之间，由应用开发者配置，每个输出批次都会形成一个RDD，以Spark作业的方式处理并生成其他的RDD。并能将处理结果按批次的方式传给外部系统。

接受器（receive）会占用一个executor的一个cpu，所以在local[n]模式下，n > 要运行的接收器数

四、DStream 操作

• DStream 上的原语与 RDD 的类似，分为 Transformations（转换，惰性的）和 Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：
  • updateStateByKey()、transform() 以及各种 Window 相关的原语。
• UpdateStateByKey 返回一个新的“状态” DStream，在该DStream中，通过在键的先前状态和键的新值上应用给定函数来更新每个键的状态。这可用于维护每个键的任意状态数据。
  • 如输入：hello world，结果则为：(hello,1)(world,1)，然后输入 hello spark，结果则为 (hello,2)(spark,1)。会保留上一次数据处理的结果。

  • import org.apache.spark.SparkConf
    
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    
    import org.apache.spark.streaming.{Durations, StreamingContext}
    
    /**
    
      * @author xiandongxie
    
      */
    
    object SparkStreamingSocketPortUpdateState {
    
      def main(args: Array[String]): Unit = {
    
        val conf: SparkConf = new SparkConf()
    
          .setAppName("sparkstreamingsocketportupdatestate")
    
          .setMaster("local[2]")
    
        val streamingContext = new StreamingContext(conf, Durations.seconds(5))
    
        // 设置保存地址
    
        streamingContext.checkpoint("/tmp/spark/sparkstreamingsocketportupdatestat")
    
        val sourceDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("localhost", 6666)
    
        val reduceDStream: DStream[(String, Int)] = sourceDStream.flatMap(_.split(" "))
    
          .map((_, 1))
    
          .reduceByKey(_ + _)
    
        val updateStateByKey: DStream[(String, Int)] = reduceDStream.updateStateByKey((newValues: Seq[Int], runningCount: Option[Int]) => {
    
          var total: Int = 0
    
          for (i <- newValues) {
    
            total += i
    
          }
    
          println(newValues  + "\t" + (if (runningCount.isDefined) runningCount.get else 0))
    
          val last: Int = if (runningCount.isDefined) runningCount.get else 0
    
          val now: Int = total + last
    
          Some(now)
    
        })
    
        updateStateByKey.foreachRDD((r, t) => {
    
          println(s"count time:${t},${r.collect().toList}")
    
        })
    
        streamingContext.start()
    
        streamingContext.awaitTermination()
    
      }
    
    }

• Transform() 原语允许 DStream 上执行任意的 RDD-to-RDD 函数 代替原有的DStream 转换操作，必须返回一个RDD；通过该函数可以方便的扩展 Spark API。

普通的转换操作如下表所示：

五、窗口转换操作

• 在Spark Streaming中，数据处理是按批进行的，而数据采集是逐条进行的，因此在Spark Streaming中会先设置好批处理间隔（batch duration），当超过批处理间隔的时候就会把采集到的数据汇总起来成为一批数据交给系统去处理。
• 对于窗口操作而言，在其窗口内部会有N个批处理数据，批处理数据的大小由窗口间隔（window duration）决定，而窗口间隔指的就是窗口的持续时间，在窗口操作中，只有窗口的长度满足了才会触发批数据的处理。除了窗口的长度，窗口操作还有另一个重要的参数就是滑动间隔（slide duration），它指的是经过多长时间窗口滑动一次形成新的窗口，滑动窗口默认情况下和批次间隔的相同，而窗口间隔一般设置的要比它们两个大。在这里必须注意的一点是滑动间隔和窗口间隔的大小一定得设置为批处理间隔的整数倍
• 

val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

窗口的计算：

Spark Streaming 还提供了窗口的计算，它允许你通过滑动窗口对数据进行转换，窗口转换操作如下：

dstream.foreachRDD { rdd =>

  rdd.foreachPartition { partitionOfRecords =>

    // ConnectionPool is a static, lazily initialized pool of connections

    val connection = ConnectionPool.getConnection()

    partitionOfRecords.foreach(record => connection.send(record))

    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse

  }

}

import org.apache.spark.SparkConf

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

import org.apache.spark.streaming.{Durations, StreamingContext}

/**

  * @author xiandongxie

  */

object SparkStreamingSocketCheckPoint {

  def main(args: Array[String]): Unit = {

    val checkpointPath = "/tmp/spark/sparkStreamingSocketCheckPoint"

    val strc: StreamingContext = StreamingContext.getOrCreate(checkpointPath, () => {

      val conf: SparkConf = new SparkConf()

        .setAppName("SparkStreamingSocketCheckPoint")

        .setMaster("local[2]")

      val streamingContext = new StreamingContext(conf, Durations.seconds(5))

      streamingContext.checkpoint(checkpointPath)

      val sourceDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("localhost", 6666)

      val reduceDStream: DStream[(String, Int)] = sourceDStream.flatMap(_.split(" "))

        .map((_, 1))

        .reduceByKey(_ + _)

      val updateStateByKey: DStream[(String, Int)] = reduceDStream.updateStateByKey((newValues: Seq[Int], runningCount: Option[Int]) => {

        var total: Int = 0

        for (i <- newValues) {

          total += i

        }

        val last: Int = if (runningCount.isDefined) runningCount.get else 0

        val now: Int = total + last

        Some(now)

      })

      updateStateByKey.foreachRDD((r, t) => {

        println(s"count time:${t},${r.collect().toList}")

      })

      streamingContext

    })

    strc.start()

    strc.awaitTermination()

  }

}

最新文章

1. 通过 listboxitem 查找属于listbox第几条数据
2. windows下安装composer抛出Composer\Downloader\TransportException异常解决办法
3. cocoapod的下载安装解释
4. Android的快速开发框架afinal
5. Android中获取电池电量
6. 购物篮模型&Apriori算法
7. Request和Response
8. PAT-013 L1-013. 计算阶乘和
9. java 新手指南
10. List总结
11. Cublas矩阵加速运算
12. 多War包合并，jetty测试
13. spring security积累
14. MySql数据库的安装和卸载
15. STL源码分析-内存分配器
16. MySQL之——如何添加新数据库到MySQL主从复制列表 【转】
17. codevs 5962 [SDOI2017]数字表格
18. VS2017更新后无法使用stdlib.h
19. 防火墙启动失败，提示最后一行出错【COMMIT】
20. Windows下安装配置免安装MySQL5.7服务器

热门文章

1. 抽签小程序，妈妈再也不用担心谁洗碗（分配任务）了，so easy
2. Python 1基础语法二(标识符、关键字、变量和字符串)
3. idle中上传jar包并使用的方法
4. word多级列表应用
5. 会话技术（session/cookie）
6. 三、ARP协议和ICMP协议
7. 05-CSV文件读取（问题）
8. Go语言 中文分词技术使用技巧（一）
9. 数论-质因数（gcd） UVa 10791 - Minimum Sum LCM
10. redis list 基本操作