【Spark】SparkStreaming入门解析（一）

（图片来源于网络，侵删）

---

一、Spark Streaming概述

【1】官网

http://spark.apache.org/streaming

【2】简介

Spark Streaming是一个基于Spark Core之上的、用于流式数据的处理实时计算框架，具有高吞吐量和容错能力强等特点

Spark Streaming和SparkCore的概念很相似，Spark Streaming使用离散化流(discretized stream)作为抽象表示，叫作DStream
DStream是随时间推移而收到的数据的序列。在内部，每个时间区间收到的数据都作为 RDD 存在，而 DStream 是由这些 RDD 所组成的序列(因此得名“离散化”)

DStream 提供了许多与 RDD 所支持的操作相类似的操作支持，还增加了与时间相关的新操作，比如滑动窗口数据输入后可以用Spark的高度抽象原语如：map、reduce、join、window等进行运算

Spark Streaming可以从很多数据源（Flume、Kafka 或者 HDFS）消费数据并对数据进行近实时处理（实质是“微批次”处理）。创建出来的DStream（InputDStream） 支持两种操作，一种是转化操作(transformation)，会生成一个新的DStream（和RDD的转换类似），另一种是输出操作(output operation)（底层实质是调用通过foreachRDD得到一个RDD来调用行动算子），可以把数据写入外部系统中(如HDFS，MySQL，Kafka等)

另外Spark Streaming也能和MLlib（机器学习）以及Graphx完美融合

【3】特点

1)易用
可以像编写离线批处理一样去编写流式程序，支持java/scala/python语言

2)容错
SparkStreaming在没有额外代码和配置的情况下可以恢复丢失的工作

3)易整合到Spark体系
流式处理与批处理和交互式查询相结合

4)高吞吐量
通过调节时间间隔，可在短时间内批量处理大量数据

---

二、Spark Streaming原理

【1】 架构

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

Spark Streaming中，会有一个接收器组件Receiver，作为一个长期运行的Task运行在一个Executor上，Receiver接收外部的数据流形成Input DStream

DStream会被按照时间间隔划分成一批一批的RDD，当批处理间隔缩短到秒级时，便可以用于处理实时数据流，时间间隔的大小可以由参数指定，一般设在500毫秒到几秒之间

对DStream进行操作就是对RDD进行操作，计算处理的结果可以传输给外部系统
Spark Streaming的工作流程像下面的图所示一样，接收到实时数据后，给数据分批次，然后传给Spark Engine（引擎）处理最后生成该批次的结果

Spark Streaming在Spark的驱动器程序—工作节点的结构的执行过程如下图所示👇

SparkStreaming为每个输入源启动对应的接收器，接收器以任务的形式运行在应用的执行器进程中，从输入源收集数据并保存为 RDD。它们收集到输入数据后会把数据复制到另一个执行器进程来保障容错性(默认行为)。数据保存在执行器进程的内存中，和缓存 RDD 的方式一样。驱动器程序中的 StreamingContext 会周期性（时间区间）地运行 Spark 作业来处理这些数据，把数据与之前时间区间中的 RDD 进行整合

注意：

StreamingContext一旦启动，对DStreams的操作就不能修改了
同一时间一个JVM中只有一个StreamingContext可以启动
stop() 方法将同时停止SparkContext，可以传入参数stopSparkContext用于只停止StreamingContext

【2】数据抽象

Spark Streaming的基础抽象是DStream(Discretized Stream，离散化数据流，连续不断的数据流)，代表持续性的数据流和经过各种Spark算子操作后的结果数据流
在内部实现上，DStream是一系列连续的RDD来表示。每个RDD含有一段时间间隔内的数据
可以从以下多个角度深入理解DStream
1）DStream本质上就是一系列时间上连续的RDD

2）对DStream的数据的进行操作也是按照RDD为单位来进行的


3）容错性
底层RDD之间存在依赖关系，DStream直接也有依赖关系，RDD具有容错性，那么DStream也具有容错性

4）准实时性/近实时性
Spark Streaming将流式计算分解成多个Spark Job，对于每一时间段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程
对于目前版本的Spark Streaming而言，其最小的Batch Size的选取在0.5~5秒钟之间
所以Spark Streaming能够满足流式准实时计算场景，对实时性要求非常高的如高频实时交易场景则不太适合

总结
简单来说DStream就是对RDD的封装，你对DStream进行操作，就是对RDD进行操作
对于DataFrame/DataSet/DStream来说本质上都可以理解成RDD

【3】DStream相关操作

DStream上的操作与RDD的类似，分为以下两种：

① Transformations(转换)

② Output Operations(输出) / Action

1）Transformations
常见Transformation—无状态转换：每个批次的处理不依赖于之前批次的数据

DStream 的转化操作可以分为无状态(stateless)和有状态(stateful)两种

【1】在无状态转化操作中，每个批次的处理不依赖于之前批次的数据。常见的 RDD 转化操作，例如 map()、filter()、reduceByKey() 等，都是无状态转化操作
无状态转化操作就是把简单的 RDD 转化操作应用到每个批次上，也就是转化 DStream 中的每一个 RDD

【2】相对地，有状态转化操作需要使用之前批次的数据或者是中间结果来计算当前批次的数据，有状态转化操作包括基于滑动窗口的转化操作和追踪状态变化的转化操作

①追踪状态变化UpdateStateByKey
UpdateStateByKey原语用于记录历史记录，有时，我们需要在 DStream 中跨批次维护状态(例如流计算中累加wordcount)。针对这种情况，updateStateByKey() 为我们提供了对一个状态变量的访问，用于键值对形式的 DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指定如何根据新的事件 更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态) 对
updateStateByKey() 的结果会是一个新的 DStream，其内部的 RDD 序列是由每个时间区间对应的(键，状态)对组成的，updateStateByKey操作使得我们可以在用新信息进行更新时保持任意的状态

②Window Operations
所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长，两者都必须是 StreamContext 的批次间隔的整数倍

2）Output/Action
Output Operations可以将DStream的数据输出到外部的数据库或文件系统
当某个Output Operations被调用时，Spark streaming程序才会开始真正的计算过程(与RDD的Action类似)

通用的输出操作 foreachRDD()，它用来对 DStream 中的 RDD 运行任意计算。这和transform() 有些类似，都可以让我们访问任意 RDD。在 foreachRDD() 中，可以重用我们在 Spark 中实现的所有行动操作。比如，常见的用例之一是把数据写到诸如 MySQL 的外部数据库中。
需要注意的：

• 连接不能写在driver层面（连接无法序列化发送到Executor，所以不能把连接创建在Driver端）
• 如果写在foreach则每个RDD都创建，得不偿失
• 增加foreachPartition，在分区创建
• 可以考虑使用连接池优化

【4】DStreams输入

【1】概述
Spark Streaming原生支持一些不同的数据源。一些“核心”数据源已经被打包到Spark Streaming 的 Maven 工件中，而其他的一些则可以通过 spark-streaming-kafka 等附加工件获取。每个接收器都以 Spark 执行器程序中一个长期运行的任务的形式运行，因此会占据分配给应用的 CPU 核心。此外，我们还需要有可用的 CPU 核心来处理数据。这意味着如果要运行多个接收器，就必须至少有和接收器数目相同的核心数，还要加上用来完成计算所需要的核心数。例如，如果我们想要在流计算应用中运行 10 个接收器，那么至少需要为应用分配 11 个 CPU 核心。所以如果在本地模式运行，不要使用local或者local[1]

【2】基本数据源

1）文件数据源

streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
Spark Streaming 将会监控 dataDirectory 目录并不断处理移动进来的文件，记住目前不支持嵌套目录

• 文件需要有相同的数据格式
• 文件进入 dataDirectory的方式需要通过移动或者重命名来实现
• 一旦文件移动进目录，则不能再修改，即便修改了也不会读取新数据

如果文件比较简单，则可以使用 streamingContext.textFileStream(dataDirectory)方法来读取文件；文件流不需要接收器，不需要单独分配CPU核

2）Socket数据源

Spark Streaming 将会监控 指定IP的某个端口并不断处理端口中的产生的新数据
streamingContext.socketTextStream

3）自定义数据源

可以通过streamingContext.receiverStream(<instance of custom receiver>)
来使用自定义的数据采集源

4）RDD队列

可以通过使用streamingContext.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD，都会作为一个DStream处理

object SparkStreaming08_QueueRDD {
  def main(args: Array[String]) {
    // 1.创建SparkConf
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("SparkStreaming08_QueueRDD")
    // 2.创建StreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(1))
    // 3.创建RDD队列
    val rddQueue = new mutable.SynchronizedQueue[RDD[Int]]()
    // 4.创建QueueInputDStream
    val inputStream = ssc.queueStream(rddQueue)
    // 5.处理队列中的RDD数据
    val mappedStream = inputStream.map(x => (x % 10, 1))
    val reducedStream = mappedStream.reduceByKey(_ + _)
    // 6.打印结果
    reducedStream.print()
    // 7.开启StreamingContext
    ssc.start()
    // 8.Create and push some RDDs into
    for (i <- 1 to 30) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }
  }
}

【5】总结

三、Spark Streaming实战

【1】WordCount

使用IDEA进行开发，Maven构建项目，添加如下依赖

<dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.2.0</version>
</dependency>

WordCount代码如下👇

object SparkStreaming01_WordCount {
  def main(args: Array[String]): Unit = {
    // 1.创建SparkConf对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming01_WordCount")
    // 2.创建StreamingContext对象，指定时间区间
    val ssc = new StreamingContext(sparkConf,Seconds(5))
    // 3.指定从socket中获取DStream数据
    val socketDStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.100.111",44444)
    // 4.对数据进行解析
    val wordToNumDStream: DStream[(String, Int)] = socketDStream.flatMap(_.split(" ")).map(_ -> 1).reduceByKey(_ + _)
    // 5.打印输出
    wordToNumDStream.print()
    // 6.开启StreamingContext
    ssc.start()
    // 7.持续等待着接收数据
    ssc.awaitTermination()
  }
}

1）我们在linux服务器上安装nc工具（nc是netcat的简称，原本是用来设置路由器,我们可以利用它向某个端口发送数据）
yum install -y nc
2）启动一个服务端并开放44444端口,等一下往这个端口发数据
nc -lk 44444
3）发送数据👇

4）查看效果👇

---

【2】updateStateByKey

在上面的WordCount案例中，我们发现每个批次的单词次数都被正确的统计出来，但是结果不能累加！
如果我们需要累加历史结果，那么我们需要使用updateStateByKey

updateStateByKey代码如下：

object SparkStreaming02_updateStateByKey {
  def main(args: Array[String]): Unit = {
    // 1.创建SparkConf对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming02_updateStateByKey")
    // 2.创建StreamingContext对象，指定时间区间
    val ssc = new StreamingContext(sparkConf,Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")
    // 3.设置checkpoint路径，用来保存历史数据，以便数据进行累加操作
    ssc.checkpoint("./updateStateByKey")
    // 4.指定从socket中获取DStream数据
    val socketDStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.100.111",44444)
    // 5.对数据进行解析
    val wordToNumDStream: DStream[(String, Int)] = socketDStream.flatMap(_.split(" ")).map(_ -> 1).updateStateByKey(sum)
    // 6.打印输出
    wordToNumDStream.print()
    // 7.开启StreamingContext
    ssc.start()
    // 8.持续等待着接收数据
    ssc.awaitTermination()
  }

  def sum(newDatas:Seq[Int],historyData:Option[Int]): Option[Int] ={
    // 1.将新数据和与历史数据相加，赋值给result
    var result = newDatas.sum + historyData.getOrElse(0)
    // 2.返回result
    Some(result)
  }
}

---

【3】 reduceByKeyAndWindow

众所周知，SparkStreaming是支持窗口操作的
滑动窗口转换操作的计算过程如下图所示


我们可以事先设定一个滑动窗口的长度，并且设定滑动窗口的时间间隔(每隔多长时间执行一次计算)

比如设置滑动窗口的长度为24H,设置滑动窗口的时间间隔为1H
那么意思就是：每隔1H计算最近24H的数据

注意：滑动窗口的长度和滑动窗口的时间间隔都必须是时间区间的整数倍，如果不是，会出现该异常👇

reduceByKeyAndWindow代码如下：

object SparkStreaming03_reduceByKeyAndWindow {
  def main(args: Array[String]): Unit = {
    // 1.创建SparkConf对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming03_reduceByKeyAndWindow")
    // 2.创建StreamingContext对象，指定时间区间
    val ssc = new StreamingContext(sparkConf,Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")
    // 3.指定从socket中获取DStream数据
    val socketDStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.100.111",44444)
    // 4.对数据进行解析
    val wordToNumDStream: DStream[(String, Int)] = socketDStream
      .flatMap(_.split(" ")).map(_ -> 1).reduceByKeyAndWindow((x:Int,y:Int)=>x+y,Seconds(10),Seconds(5))
    // 5.打印输出
    wordToNumDStream.print()
    // 6.开启StreamingContext
    ssc.start()
    // 7.持续等待着接收数据
    ssc.awaitTermination()
  }
}

---

【4】 统计一定时间内的热门词汇TopN

需求：模拟百度热搜排行榜

统计最近10s的热搜词Top3，每隔5秒计算一次
WindowDuration = 10s
SlideDuration = 5s

object SparkStreaming04_TopN {
  def main(args: Array[String]): Unit = {
    // 1.创建SparkConf对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming04_TopN")
    // 2.创建StreamingContext对象，指定时间区间
    val ssc = new StreamingContext(sparkConf,Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")
    // 3.指定从socket中获取DStream数据
    val socketDStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.100.111",44444)
    // 4.对数据进行解析
    val wordToNumDStream: DStream[(String, Int)] = socketDStream.flatMap(_.split(" ")).map(_ -> 1).reduceByKeyAndWindow((x:Int,y:Int)=>x+y,Seconds(10),Seconds(5))
    // 5.将DStream转换为RDD，对RDD进行操作，取出Top3
    val sorteDStream: DStream[(String, Int)] = wordToNumDStream.transform(rdd => {
      val sortedRDD: RDD[(String, Int)] = rdd.sortBy(_._2, false) //逆序/降序
      println("===============top3==============")
      sortedRDD.take(3).foreach(println)
      println("===============top3==============")
      sortedRDD
      }
    )
    // 6.打印数据
    sorteDStream.print
    // 7.开启StreamingContext
    ssc.start()
    // 8.持续等待着接收数据
    ssc.awaitTermination()
  }
}

---

都看到这里了，点赞评论一下吧！！！

点击查看👇

【Spark】SparkStreaming入门解析（二）

转载：https://blog.csdn.net/qq_43733123/article/details/105441252