返回 登录
2

Spark-Streaming的基本原理以及预写日志机制和checkpoint,面试大杀器!!!!

首先我们要先了解什么是Spark-Streaming:
Spark Streaming是Spark Core API的一种扩展,它可以用于进行大规模、高吞吐量、容错的实时数据流的处理。它支持从很多种数据源中读取数据,比如Kafka、Flume、Twitter、ZeroMQ、Kinesis或者是TCP Socket。并且能够使用类似高阶函数的复杂算法来进行数据处理,比如map、reduce、join和window。处理后的数据可以被保存到文件系统、数据库、Dashboard等存储中。

接下来要知道Spark-Streaming的基本运行原理:
Spark-Streaming内部的基本工作原理如下:接收实时输入数据流,然后将数据拆分成多个batch,比如每收集5秒的数据封装为一个batch,然后将每个batch交给Spark的计算引擎进行处理,最后会生产出一个结果数据流,其中的数据,也是由一个一个的batch所组成的。

关于Spark-Streaming的高级抽象
Spark-Streaming提供了一种高级的抽象,叫做DStream,英文全称为Discretized Stream,中文翻译为“离散流”,它代表了一个持续不断的数据流。DStream可以通过输入数据源来创建,比如Kafka、Flume和Kinesis;也可以通过对其他DStream应用高阶函数来创建,比如map、reduce、join、window。
DStream的内部,其实一系列持续不断产生的RDD。RDD是Spark Core的核心抽象,即,不可变的,分布式的数据集。DStream中的每个RDD都包含了一个时间段内的数据。

接下来就是一些关于面试时的大杀器:

Spark-Streaming checkPoing概述

每一个Spark Streaming应用,正常来说,都是要7*24小时运转的,这就是实时计算程序的特点,因为要持续不断地对数据进行计算,因此,对实时计算应用的要求,应该是必须要能够对应用程序逻辑无关的失败,进行容错,如果要实现这个目标,Spark-Streaming程序就必须讲座狗的信息checkpoint到容错的存储系统上,从而让它能够错失败中进行恢复

如何对dstream做checkpoint?
首先设置还原点目录,其次调用dstream的checkpoint方法
【注意】:dstream的checkpoint的周期一定要是产生batch的时间的整数倍,同时官方建议将checkpoint的事件设置为至少10秒,
通常来说,将checkpoint间隔设置为窗口操作的滑动间隔的5~10倍是个不错的选择

有两种数据需要被进行checkpoint:

1.元数据checkpoint—-将定义了流式计算逻辑的信息,报错到容错的存储系统上,比如HDFS
当运行Spark—Streaming应用程序的Driver进程所在的节点失败时,该信息可以用于进行恢复。
元数据信息包括了:
1.1:配置信息—创建Spark-Streaming应用程序的配置信息,比如SparkConf
1.2:DStream的操作信息—-定义了Spark-Stream应用程序的计算逻辑的DStream操作信息
1.3:未处理的batch信息—-哪些job正在排队,还没处理的batch信息。

2.数据checkpoint—将实时计算过程中产生的RDD的数据保存到可靠的存储系统中
对于一些将多个batch的数据进行聚合的,有状态的transformation操作,这是非常有用的,
在这种tranformation操作中,生成的RDD是依赖与之前的batch的,这会导致随着时间的推移,Rdd的依赖
链条越来越长,要避免由于依赖链条越来越长,导致一起变得越来越长的失败恢复时间,有状态的transformation
操作执行过程中间产生的RDD,会定期的被checkpoint盗可靠的存储系统上,比如HDFS,从而削减RDD的依赖链条,进而缩短失败恢复时,
RDD的回复时间

合适启用checkpoint机制
1.使用了有状态的transformation操作—比如updateStateByKey,或者reduceByKeyAndWindow操作被使用了,
那么checkpoint目录要求是必须提供的,也就必须开启checkpoint机制,从而进行周期性的RDD checkpoint
2.要保证可以从Driver失败中进行恢复—元数据checkpoint需要启用,来进行这种情况的恢复,
要注意的是,并不是说所有的Spark-Streaming应用程序,都要启用checkpoint机制,如果不强制要求从Driver
失败中自动进行恢复,有没有使用有状态的transformation操作,那么就不需要启用checkpoint,事实上
这么做反而是用利于提升性能的。

启动预写日志机制
预写日志机制,简写为WAL,全称为Write Ahead Log,从Spark 1.2版本开始,就引入了基于容错的文件系统的WAL机制,如果启用该机制,Receiver接收到的所有数据都会被写入配置的checkpoint目录中的预写日志这种机制可以让driver在恢复的时候,避免数据丢失,并且可以确保整个实时计算过程中,零数据丢失

要配置该机制,首先要调用StreamingContext的checkpoint()方法设置一个checkpoint目录,然后需要将spark.streaming.receiver.writeAheadLog.enable参数设置为true

然而这种极强的可靠性机制,会导致Receiver的吞吐量大幅度下降,因为单位时间内有相当一部分时间需要将数据写入预写日志,如果又希望开启预写日志机制,确保数据零损失,又不希望影响系统的吞吐量,那么可以创建多个输入DStream启动多个Reciver,然后将这些receiver接收到的数据使用ssc.union()方法将这些dstream中的数据进行合并
此外在启用了预写日志机制之后,推荐将复制持久化机制禁用掉,因为所有数据已经保存在容错的文件系统中了,不需要再用复制机制进行持久化,保存一份副本了,只要将输入的DStream的持久化机制设置一下即可

评论