资讯

历史

科技

环境与自然

成长

游戏

财经

文学与艺术

美食

健康

家居

文化

情感

汽车

三农

军事

旅行

运动

教育

生活

星座命理

Spark Streaming 原理总结：一个大数据流式数据（准实时）计算框架

创作时间:

作者:

@小白创作中心

Spark Streaming 原理总结：一个大数据流式数据（准实时）计算框架

引用

CSDN

https://blog.csdn.net/The_skye/article/details/142576005

Spark Streaming是Apache Spark生态系统中的重要组件，用于处理实时流式数据。本文将从Spark Streaming的基本概念、核心原理、数据处理流程以及应用场景等多个维度进行深入探讨，帮助读者全面理解这一强大的流式数据处理框架。

Spark Streaming原理总结

1.1、 Spark Streaming概念

Spark Streaming 是一个流式数据（准实时）计算框架。其基本原理是把输入数据以某一时间间隔批量的处理，便可以用于处理实时数据流。其本质是批处理，将流式数据抽象成一个一个的批次，当批次小到一定程度时（微批），便可视为处理实时数据流。

准实时：区别于离线和实时，准实时介于二者之间。通常情况下，离线开发延迟一般以天、小时为单位，实时开发延迟在毫秒级别；而准实时则是秒或者分钟级别。

1.2、DStream：离散化流

Discretized Stream是Spark Streaming的基础抽象（和Spark 基于RDD 的概念很相似）代表持续性的数据流和经过各种Spark原语操作后的结果数据流。在内部实现上，DStream是一系列连续的RDD来表示。每个RDD含有一段时间间隔内的数据。

所以简单来讲，DStream就是对RDD在实时数据处理场景的一种封装。

此时，Spark计算框架处理流式数据也就转化成了对RDD的处理。

2、Spark Streaming处理流式数据的过程

Apache Spark Streaming 是一个实时流处理框架，它在处理数据时会将数据分成一系列连续的批次，然后对这些批次进行处理。在 Spark Streaming 中，数据的接收、转换和输出是流处理的三个主要步骤。

上图中展示了流式数据通过Spark Streaming计算框架转化成一系列（微批）数据，并通过Spark Core计算引擎计算后输出一系列处理过的数据。

由Spark Streaming架构图可以看出，流式数据最终还是使用Spark Context来做计算，也就是说底层还是对RDD（分布式弹性数据集）来做计算。

流式数据WordCount demo：

package spark.streaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    // 1. 初始化Spark配置信息
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamWordCount")
    // 2. 初始化SparkStreamingContext 设置批次大小
    val ssc: StreamingContext = new StreamingContext(conf = conf, batchDuration = Seconds(3))
    // 3. 通过监控端口创建DStream，读进来的数据为一行行
    val lineStreams: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    // 将每一行数据做切分，形成一个个单词 并将单词映射成元组（word,1）
    val wordStream: DStream[(String, Int)] = lineStreams.flatMap(_.split(" ")).map((_, 1))
    // 做词频统计
    val wordCountStream: DStream[(String, Int)] = wordStream.reduceByKey(_ + _)
    // print
    wordCountStream.print()
    // start SparkStreamContext and waiting for termination
    ssc.start()
    ssc.awaitTermination()
  }
}

具体计算过程：

数据接收：Spark Streaming 通过接收器（Receiver）从各种数据源（如 Kafka、Flume、Kinesis 或 TCP 套接字）接收实时数据流。接收器负责将数据从源头拉取到 Spark Streaming 应用程序中，并将其存储在 Spark 的内存中以供处理。
数据处理：接收到的数据会被分成批次，每个批次被表示为一个弹性分布式数据集（RDD）。Spark Streaming 提供了丰富的转换（算子）操作（如 map、filter、reduceByKey 等），允许用户对数据进行复杂的转换和处理。
数据输出：处理后的数据可以被推送到各种外部系统中，如文件系统、数据库或实时看板。输出操作负责将数据从 Spark Streaming 应用程序传输到外部系统。

3、 DStream转换

DStream 上的操作与RDD的类似，分为Transformations（转换）和Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及各种Window相关的原语。

3.1、无状态转化操作

无状态转化操作就是把简单的RDD转化操作应用到每个批次上，也就是转化DStream中的每一个RDD。部分无状态转化操作列在了下表中。

需要记住的是，尽管这些函数看起来像作用在整个流上一样，但事实上每个DStream在内部是由许多RDD（批次）组成，且无状态转化操作是分别应用到每个RDD上的。

例如：reduceByKey()会归约每个时间区间中的数据，但不会归约不同区间之间的数据。

3.1.1 Transform

Transform 允许 DStream 上执行任意的RDD-to-RDD函数。即使这些函数并没有在DStream的API中暴露出来，通过该函数可以方便的扩展Spark API。该函数每一批次调度一次。其实也就是对DStream中的RDD应用转换。

3.1.2 join

两个流之间的join需要两个流的批次大小一致，这样才能做到同时触发计算。计算过程就是对当前批次的两个流中各自的RDD进行join，与两个RDD的join效果相同。

3.2、有状态转化操作

WindowOperations

Window Operations可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态。所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。

➢ 窗口时长：计算内容的时间范围；
➢ 滑动步长：隔多久触发一次计算。

注意：这两者都必须为采集周期大小的整数倍。且进行有状态转化操作需要对检查点（checkpoint）目录进行配置，会使用检查点来保存状态。

WordCount示例：

package spark.streaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object WindowOperations {
  // Window Operations可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态。
  // 所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。
  //    窗口时长：计算内容的时间范围；
  //    滑动步长：隔多久触发一次计算。 (这两者都必须为采集周期大小的整数倍)
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("WindowOperations")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 必须设置检查点
    ssc.checkpoint("./ck")
    // Create a DStream that will connect to hostname:port, like localhost:9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    // Split each line into words
    val words: DStream[String] = lines.flatMap(_.split(" "))
    // Count each word in each batch
    val pairs: DStream[(String, Int)] = words.map(word => (word, 1))
    val wordCounts: DStream[(String, Int)] = pairs.reduceByKeyAndWindow(
      // 加上新进入窗口的批次中的元素
      (a, b) => a + b,
      // 移除离开窗口的老批次中的元素
      (a, b) => a - b,
      // 窗口时长
      Seconds(12),
      // 滑动步长
      Seconds(6)
    )
    // Print the first ten elements of each RDD generated in this DStream to the console
    wordCounts.print()
    // Start the computation and wait for termination
    ssc.start()
    ssc.awaitTermination()
  }
}