Fork me on GitHub

Spark Streaming 实时计算框架

字数统计: 2.9k字   |   阅读时长≈ 11分

Spark Streaming 实时计算框架

有时候需要通过淘宝、京东等网站从用户点击行为（如加入购物车）和浏览的记录中发现用户的购买意图和兴趣，然后通过 Spark Streaming 实时计算框架分析处理，为之推荐相关商品，有效地提高商品销量。

实时计算框架

实时计算框架可以针对海量数据进行实时计算，无论在数据采集还是数据处理中，都可以达到秒级处理。

常用框架有：

  • Apache Spark Streaming, Apache 免费、开源的实时计算框架。把输入的数据按时间拆分，并对切分的数据块进行并行计算处理。Netflix 公司通过 Kafka 和 Spark Streaming 构建了实时引擎，每天从各种数据源接收到的数十亿数据进行分析，从而完成电影推荐功能。
  • Apache Storm, Twitter 开源贡献给 Apache ， 是一个分布式实时计算系统。开源简单、高效、可靠地实时处理海量数据，处理数据速度达到毫秒级，同时开源将处理的结果数据持久化（数据库、HDFS）。阿里巴巴的 JStorm 计算参考 Apache Storm 开发，在网络 IO、线程模型、资源调度、可用性和稳定性都做了改进，是Storm的增强版本.
  • Apache Flink, Apache 开源的计算框架。不仅可以离线处理，还可以支持实时处理。由于离线处理和实时处理所提供的 SLA（服务等级协议）是完全不相同的，所以离线处理一般需要支持帝延迟的保证，而实时处理则需要支持高吞吐、高效率的处理。
  • Yahoo! S4（Simple Scalable Streaming System），Yahoo 公司开源的实时计算平台。是通用的、分布式的、可扩展的，并且具有容错和可插拔能力，供开发者轻松地处理源源不断产生的数据。

Spark Streaming 基础

Spark Streaming 是构建在 Spark 的实时计算框架，是 Spark Core API 的一个扩展，能够实现对流数据进行实时处理，具有可扩展、高吞吐量和容错性。

Spark Streaming 支持从多种数据获取数据，包括 Kafka、Flume、Twitter、ZeroMQ、Kinesis 以及 TCP Socket 数据源。当获取到数据后，可以使用 map、reduce、join 和 window 等高级函数进行复杂的计算处理，最后将处理结果存储到分布式文件系统、数据库中，利用实时 Web 仪表盘(Dashboards)进行展示。

工作原理：Spark Streaming 接收实时数据流，将数据按照一定时间间隔分成多批数据，每一段数据转变成 Spark RDD，接着交由 Spark 引擎进行处理，最后将处理结果数据输出到外部存储系统。

  1. 易用性。支持 Java、Python、Scala 等编程语言，像操作离线程序一样编写实时计算程序。
  2. 容错性。Spark Streaming 没有额外代码和配置的情况可以恢复丢失的数据。Spark 中的 RDD 的容错机制，每个 RDD 都是一个不可变的分布式可重算的数据集，它记录着确定性的操作继承关系（lineage），所以主要输入数据是可容错的，任意一个 RDD 的分区（Partition）出错或不可用都可以使用原始输入数据经过转换操作重新计算得到。
  3. 易整合性。Spark Streaming 可以在 Spark 上运行，允许重复使用代码进行批处理。可以实时处理与离线处理相结合，实现交互式查询操作。

DStream

Spark Streaming 核心是 DStream（离散流），是高级抽象的流。DStream 表示连续的数据流，可以通过 Kafka、Flume 和 Kinesis 等数据源插件，也可以通过现有的高级操作来创建。DStream 的内部结构是由一系列连续的 RDD 组成，每个 RDD 都是一小段由世界分隔开来的数据集。对 DStream 的所有操作，都会转变成底层 RDD 的操作。

DStream API 相关方法：

方法名称 相关说明
map(func) 将源DStream的每个元素，传递到函数func进行转换操作得到新的DStream
flatMap(func) map()相似，每个输入元素可映射 0 或者多个输出结果
filter(func) 返回新的DStream，仅包含DStream中经过func函数计算结果为true的元素
repartition(numPartitions) 指定DStream分区数量
union(otherStream) 返回新的DStream，包含源DStream和其他DStream中所有元素
count() 统计源DStream每个RDD包含的元素个数，返回一个新的DStream
reduce(func) func（两个参数并返回一个结果）将源DStream中每个RDD的元素进行聚合操作，返回新的DStream
countByValue() 计算DSteram每个RDD内的元素出现的频次返回新的DStream[(K,Long)],K是RDD的元素类型，Long是元素出现频次
reduceByKey(func,[numTasks]) 当类型为(K,V)的DStream被调用是，则返回一个类型为(K,V)的新DStream，每个键的值都是使用聚合函数func汇总得到。注意：默认并行度为2，集群模式并行度为8，可通过配置参数 numTasks 来设置不同的并行任务数
join(otherStream,[numTasks]) 调用类型分别为(K,V)和(K,V)的两个DStream是返回类型为(K,(V,W))的新的DStream
cogroup(otherStream,[numTasks]) 调用类型分别为(K,V)和(K,V)的两个DStream是返回类型为(K,Seq[V],Seq[W])类型的新DStream
transfrom(func) 通过源DStream中的每个 RDD应用RDD-to-RDD函数返回一个新的DStream，这样可以在DStream中做任意RDD操作
updateStateByKey(func) 返回新状态的DStream，通过键的先前状态和间的新值上应用给定函数func来更新每个键的状态。该方法主要用于维护每个键任意状态数据。

对比RDD API，RDD API没有提供 transform() 和 updateStateByKey()

使用 transfrom() 将一行语句分割成多个单词。

1
2
3
4
5
# 启动服务端且监听 Socket 服务（Socket 服务端口为 9999）
nc -lk 9999
# 运行 scala 程序后
# 输入单词
I am learning Spark Streaming now

添加依赖，配置 pom.xml 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
<!-- 设置依赖版本号 -->
<properties>
    <!-- 对应spark-core_2.12 -->
    <scala.version>2.12.7</scala.version>
    <hadoop.version>2.10.1</hadoop.version>
    <spark.version>2.4.2</spark.version>
</properties>
<dependencies>
    <!--scala-->
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>${scala.version}</version>
    </dependency>
    <!--Spark-->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>${spark.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.12</artifactId>
        <version>${spark.version}</version>
    </dependency>
</dependencies>
<build>
    <sourceDirectory>src/main/scala</sourceDirectory>
    <testSourceDirectory>src/test/scala</testSourceDirectory>
</build>

创建 TransformTest.scala 类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object TransformTest {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf()
      .setAppName("TransformTest").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")

    // 创建 StreamingContext ，需要两个参数，分别为 SparkContext 和批处理时间间隔
    val ssc:StreamingContext = new StreamingContext(sc, Seconds(5))
    // 链接 socket 服务，需要 socket 服务地址、端口号以及存储级别
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.124.135", 9999)
    // 使用 RDD-to-RDD 函数，返回新的 DStream 对象（即words），并空格切分每行
    val words: DStream[String] = dstream.transform(rdd => rdd.flatMap(_.split(" ")))
    // 打印结果
    words.print()
    // 开启流式运算
    ssc.start()
    // 保持程序运行
    ssc.awaitTermination()
  }
}

使用 updateStateByKey() 进行词频统计，创建 UpdateStateByKeyTest.scala 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
object UpdateStateByKeyTest {

  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]) : Option[Int] = {

    val newCount = runningCount.getOrElse(0) + newValues.sum
    Some(newCount)
  }

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

    val sparkConf: SparkConf = new SparkConf()
      .setAppName("UpdateStateByKeyTest").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")

    // 创建 StreamingContext ，需要两个参数，分别为 SparkContext 和批处理时间间隔
    val ssc:StreamingContext = new StreamingContext(sc, Seconds(5))
    // 配置检查点点目录，使用 updateStateByKey 方法必须配置检查点目录
    ssc.checkpoint("./")
    // 链接 socket 服务，需要 socket 服务地址、端口号以及存储级别
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.124.135", 9999)

    // 使用 RDD-to-RDD 函数，返回新的 DStream 对象（即words），并空格切分每行
    val wordAndOne: DStream[(String, Int)] = dstream.flatMap(_.split(" ").map(word => (word, 1)))
    // 调用 updateStateByKey 操作，统计单词在全局中出现的次数
    val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunction)
    // 打印结果
    result.print()
    // 开启流式运算
    ssc.start()
    // 保持程序运行
  }
}

每隔 5s 接收一次数据，一共接收到多次数据。

DStream 窗口操作

窗口操作即将一个可配置的长度设置为窗口，以一个可配置的速率向前移动。根据窗口操作，对窗口内的数据进行计算，每次落在窗口内的 RDD 数据会被聚合起来计算，生产的 RDD 会作为 window DStream 的一个 RDD。

方法名称 相关说明
window(windowLength,slideInterval) 返回基于源DStream的窗口进行批计算后的新的DStream
countByWindow(windowLength,slideInterval) 返回基于滑动窗口的DStream中的元素
reduceByWindow(func,windowLength,slideInterval) 基于滑动窗口的源DStream 中的元素进行聚合操作，返回一个新的 DStream
reduceByKeyAndWindow(func,windowLength,slideInterval,[numTasks]) （K,V）类型按 K 应用聚合函数 func 进行聚合操作，返回新DStream
reduceByKeyAndWindow(func,invFuncwindowLength,slideInterval,[numTasks]) 更高效的 本版本。每个窗口聚合值都基于先前窗口的聚合值进行增量计算得到。操作会对进入活动窗口的新数据进行聚合操作，并对离开窗口的历史数据进行逆向聚合操作。
countByValueAndWindow(windowLength,slideInterval,[numTasks]) 计算源DStream 中每个 RDD 内每个元素出现的频次，返回 （K,V）类型的DSteram，K为RDD 中的元素类型，V为元素在滑动窗口出现的次数

window 输出3个时间单位长度的时间 WindowTest.scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
object WindowTest {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf()
      .setAppName("UpdateStateByKeyTest").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")

    // 创建 StreamingContext ，需要两个参数，分别为 SparkContext 和批处理时间间隔
    val ssc:StreamingContext = new StreamingContext(sc, Seconds(1))
    // 链接 socket 服务，需要 socket 服务地址、端口号以及存储级别
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.124.135", 9999)
    // 按空格切分每一行
    val words: DStream[String] = dstream.flatMap(_.split(" "))
    // 调用 window 操作，需要两个参数，窗口长度和滑动时间间隔
    val windowWords: DStream[String] = words.window(Seconds(3), Seconds(1))
    // 打印输出结果
    windowWords.print()
    // 开启流式计算
    ssc.start()
    // 让程序一直运行
    ssc.awaitTermination()
  }
}

使用 reduceByKeyAndWindow() 方法统计3个时间单位内不同字母出现的次数。 ReduceByKeyAndWindowTest.scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
object ReduceByKeyAndWindowTest {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf()
      .setAppName("UpdateStateByKeyTest").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")

    // 创建 StreamingContext ，需要两个参数，分别为 SparkContext 和批处理时间间隔
    val ssc:StreamingContext = new StreamingContext(sc, Seconds(1))
    // 链接 socket 服务，需要 socket 服务地址、端口号以及存储级别
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.124.135", 9999)
    // 按空格切分每一行
    val words: DStream[(String, Int)] = dstream.flatMap(_.split(" ").map(word => (word, 1)))
    // 调用 reduceByKeyAndWindow 操作
    val windowWords: DStream[(String, Int)] = words.reduceByKeyAndWindow((a:Int, b:Int) => a+b, Seconds(3), Seconds(1))
    // 打印输出结果
    windowWords.print()
    // 开启流式计算
    ssc.start()
    // 让程序一直运行
    ssc.awaitTermination()
  }
}
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
后端开发、微服务2.0(服务网格)
、大数据、人工智能、超级账本
打工仔、开源爱好者、极客
联系QQ:2595903671
很惭愧
只做了一点微小的工作