其他分享
首页 > 其他分享> > 第10章《Spark Streaming》

第10章《Spark Streaming》

作者:互联网

目录

1 简介

许多应用需要即时处理收到的数据,例如用来实时追踪页面访问统计的应用、训练机器学习模型的应用,还有自动检测异常的应用。Spark Streaming是Spark为这些应用而设计的模型。Spark Streaming使用离散化流作为抽象表示,叫做DStream。DStream可以从各种输入源创建,比如Flume、Kafka或者HDFS。创建出来的DStream支持两种操作,一种是转化操作(transformation),会生成一个新的DStream,另外一种是输出操作(output operation),可以把数据写入外部系统中

2 实例

让我们先来看一个简单的例子。我们会从一台服务器的7777端口上收到一个以换行符分隔的多行文本,要从中筛选出包含单词 error 的 行,并打印出来。Spark Streaming 程序最好以使用 Maven 或者 sbt 编译出来的独立应用的形式运行。Spark Streaming 虽然是 Spark 的一部分,它在 Maven 中也以独立工件的形式提供,你也需要在 工程中添加一些额外的 import 声明。
我们从创建 StreamingContext开始,它是流计算功能的主要入口。StreamingContext 会在底层创建出SparkContext,用来处理数据。其构造函数还接收用来指定多长时间处理一次新数据的批次间隔(batch interval)作为输入,这里我们把它设为1秒。接着,调 用socketTextStream() 来创建出基于本地 7777 端口上收到的文本数据的 DStream。然后把 DStream通过filter() 进行转化,只得到包含“error”的行。最后,使用输出操作 print() 把一些筛选出来的行打印出来。如下是基于Scala进行流式筛选操作:

val ssc = new StreamingContext(conf, Second(1))
val lines = ssc.socketTextStream("localhost", 7777)
val errorLines = lines.filter(_.contains("error"))
errorLines.print()

。要开始接收数据,必须显式调用 StreamingContext的start()方法。这样,Spark Streaming 就会开始把Spark作业不断交给下面的 SparkContext 去调度执行。执行会在另一个线程中进行,所以需要调用awaitTermination 来等待流计算完成,来防止应用退出。

ssc.start()
ssc.awaitTermination()

3 架构与抽象

Spark Streaming 使用“微批次”的架构,把流式计算当作一系列连续的小规模批处理来对待。Spark Streaming 从各种输入源中读取数据,并把数据分组为小的批次。新的批次按均匀的时间间隔创建出来。在每个时间区间开始的时候,一个新的批次就创建出来,在该区间内收到的数据都会被添加到这个批次中。在时间区间结束时,批次停止增长。每个输入批次都形成一个RDD,以 Spark 作业的方式处理并生成其他的 RDD
高层次的架构图如下:
在这里插入图片描述
我们已经讲到过,Spark Streaming的编程抽象是离散化流,也就是 DStream。它是一个 RDD序列,每个 RDD 代表数据流中一个时间片内的数据。如下图所示:
在这里插入图片描述
你可以从外部输入源创建 DStream,也可以对其他 DStream 应用进行转化操作得到新的DStream。DStream 支持许多RDD的转化操作。另外,DStream还有 “有状态”的转化操作,可以用来聚合不同时间片内的数据。上述例子中的DStream及其转化关系如下图所示:
在这里插入图片描述
除了转化操作以外,DStream还支持输出操作,比如在示例中使用的 print()。输出操作 和 RDD 的行动操作的概念类似。Spark 在行动操作中将数据写入外部系统中,而 Spark Streaming 的输出操作在每个时间区间中周期性执行,每个批次都生成输出
Spark Streaming 对DStream 提供的容错性与 Spark为RDD 所提供的容错性一致:只要输入数据还在,它就可以使用 RDD 谱系重算出任意状态 (比如重新执行处理输入数据的操 作)。默认情况下,收到的数据分别存在于两个节点上,这样 Spark 可以容忍一个工作节点的故障。不过,如果只用谱系图来恢复的话,重算有可能会花很长时间,因为需要处理从 程序启动以来的所有数据。因此,Spark Streaming 也提供了检查点机制,可以把状态阶段 性地存储到可靠文件系统中(例如 HDFS或者S3)。一般来说,你需要每处理 5-10 个批次 的数据就保存一次。在恢复数据时,Spark Streaming 只需要回溯到上一个检查点即可。

4 转化操作

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

5 输出操作

输出操作指定了对流数据经转化操作得到的数据所要执行的操作(例如把结果推入外部数据库或输出到屏幕上)。与RDD中的惰性求值类似,如果一个 DStream 及其派生出的 DStream 都没有被执行输出操作,那么这些 DStream 就都不会被求值。如果 StreamingContext 中没有设定输出操作,整个context 就都不会启动。常用的一种调试性输出操作是print(),它会在每个批次中抓取 DStream 的前十个元素打印出来。Spark Streaming 对于 DStream 有 与 Spark 类似的 save() 操作,它们接受一个目录作为参数来存储文件,还支持通过可选参数来设置文件的后缀名。Spark Streaming 对于 DStream 有 与 Spark 类似的 save() 操作,它们接受一个目录作为参数来存储文件,还支持通过可选参数来设置文件的后缀名。在Scala中将DStream保存为文本文件如下:

requestCount.saveAsTextFiles("outputDir","txt")

还有一个更为通用的saveAsHadoopFiles() 函数,接收一个 Hadoop 输出格式作为参数。最后,还有一个通用的输出操作 foreachRDD(),它用来对 DStream 中的 RDD 运行任意计算。这和 transform() 有些类似,都可以让我们访问任意 RDD。在foreachRDD() 中,可以重用我们在Spark 中实现的所有行动操作。foreachRDD() 也可以提供给我 们当前批次的时间,允许我们把不同时间的输出结果存到不同的位置。

6 24/7不间断运行

Spark Streaming 的一大优势在于它提供了强大的容错性保障。只要输入数据存储在可靠的系统中,Spark Streaming 就可以根据输入计算出正确的结果,提供“精确一次”执行的语义。检查点机制是我们在Spark Streaming 中用来保障容错性的主要机制。它可以使 Spark Streaming 阶段性地把应用数据存储到诸如 HDFS 或 Amazon S3 这样的可靠存储系统中, 以供恢复时使用。检查点机制主要为以下两个目的服务:

标签:10,RDD,批次,Streaming,操作,Spark,DStream
来源: https://blog.csdn.net/BGoodHabit/article/details/122573518