spark streaming ,预写日志（write-ahead logging）和checkpoint。

我会带着你远行 2022-05-26 07:58 142阅读 0赞

# （1）什么是Spark-Streaming？ #

spark steaming 是spark Core API的一种扩展，它可用于大规模、高吞吐量、容错的实时数据流处理。它支持从多种数据源中读取数据，如 kafka,flume,twitter,zeromq,kinesis 或者是tcp socket。并且能够使用类似高阶函数的复杂算法来进行数据处理，如 map(),reduce(),join(),window(）等。处理后的数据可以保存到文件系统、数据库等。

# （2）spark streaming的基本运行原理？ #

接收实时输入数据流，将数据流按照时间间隔拆分成多个batch,比如每收集5s的数据封装成一个batch,然后将每个batch交给spark的计算引擎进行处理，产生由一个一个的batch组成的结果数据流，输出。

# （3）spark streaming的高级抽象？ #

spark streaming提供的高级抽象叫做Dstream，即 Discretized Stream(离散流)  
代表了一个持续不断的数据流，Dstream可以利用数据源来创建，如kafka,flume,twitter,zeromq,kinesis，也可以通过其他的Dstream通过高阶函数来创建，如map(),reduce(),join(),window(）等，自我的理解是类似于RDD的创建，可以根据数据本身来创建，也可以根据RDD通过map(),reduce(),join()等来创建，Dstream类似于一个装满RDD元素的RDD列表的感觉。Dstream的内部其实是一系列持续不断产生的RDD，RDD是不可变的分布式的数据集，Dstream 中的每个RDD都包含了一个时间段内的数据。

# （4）spark streaming checkpoint的概述？ #

每一个spark streaming应用程序，都需要持续不断的进行运转计算，不可停歇，因此就必须有对与应用逻辑无关的失败的容错能力，spark streaming就要将错的信息checkpoint在容错的存储系统上，是应用程序运行时，能够从错败中恢复。

# （5）如何对Dstream checkpoint? #

首先，设置还原点目录；  
其次，调用Dstream的checkpoint方法；  
Dstream的checkpoint周期一定要是产生batch的整数倍，官方建议设置至少10s,通常来说设置为窗口滑动间隔的5~10倍比较合适。

# （6）spark streaming过程中，设么数据要被checkpoint？ #

（1）元数据信息。  
包括：  
（1.1）配置信息—创建Spark-Streaming应用程序的配置信息，比如SparkConf  
（1.2）DStream的操作信息—-定义了Spark-Stream应用程序的计算逻辑的DStream操作信息  
（1.3）未处理的batch信息—-哪些job正在排队，还没处理的batch信息。  
（2）实时计算中产生的RDD数据checkpoint。对于一些将多个batch的数据进行聚合的，有状态的transformation操作，这是非常有用的，  
在这种tranformation操作中，生成的RDD是依赖与之前的batch的，这会导致随着时间的推移，Rdd的依赖  
链条越来越长，要避免由于依赖链条越来越长，导致一起变得越来越长的失败恢复时间，有状态的transformation  
操作执行过程中间产生的RDD，会定期的被checkpoint盗可靠的存储系统上,比如HDFS,从而削减RDD的依赖链条，进而缩短失败恢复时RDD的恢复时间。

# （7）什么情况下启用checkpoint机制比较合适？ #

（1）使用了有状态的transformation操作—比如updateStateByKey,或者reduceByKeyAndWindow操作被使用了，  
那么checkpoint目录要求是必须提供的，也就必须开启checkpoint机制，从而进行周期性的RDD checkpoint 。  
（2）保证driver能够从失败中恢复，元数据的checkpoint需要启动，来应对driver失败，但如果没有强制要求从Driver失败中自动恢复，又没有使用有状态的transformation 操作，那么就不需要启用checkpoint。

# （8）预写日志是什么？其工作机制是什么？如何配置预写日志机制？ #

预写日志，简称WAL,从spark1.2版本开始，引入基于容错机制的文件系统的WAL机制，若启用该机制，receiver接收到的所有数据都会被写入配置的checkpoint目录中的预写日志这种机制可以让driver在恢复的时候，避免数据丢失，并且可以确保整个实时计算过程中，零数据丢失。  
要配置该机制：  
首先，调用StreamingContext的checkpoint()方法设置一个checkpoint目录；  
然后，将spark.streaming.receiver.writeAheadLog.enable参数设置为true。

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