容错和持久化：保障系统稳定性

1.容错

DStream基于RDD组成，RDD的容错性依旧有效，我们首先回忆一下SparkRDD的基本特性。

（1）RDD是一个不可变的、确定性的可重复计算的分布式数据集。RDD的某些parti-tion丢失了，可以通过血统（Lineage）信息重新计算恢复。

（2）如果RDD任何分区因Worker结点故障而丢失，那么这个分区可以从原来依赖的容错数据集中恢复。

（3）由于Spark中所有的数据的转换操作都是基于RDD的，即使集群出现故障，只要输入数据集存在，所有的中间结果都是可以被计算的。

Spark Streaming是可以从HDFS和S3这样的文件系统读取数据的，这种情况下所有的数据都可以被重新计算，不用担心数据的丢失。但是在大多数情况下，Spark Streaming是基于网络来接受数据的，此时为了实现相同的容错处理，在接受网络的数据时会在集群的多个Worker结点间进行数据的复制（默认的复制数是2），这导致产生在出现故障时被处理的两种类型的数据：

●Data received and replicated：一旦一个Worker结点失效，系统会从另一份还存在的数据中重新计算。

●Data received but buffered for replication：一旦数据丢失，可以通过RDD之间的依赖关系，从HDFS这样的外部文件系统读取数据。

此外，有两种故障，我们应该关心：

●Worker结点失效：通过上面的讲解我们知道，这时系统会根据出现故障的数据的类型，选择是从另一个有复制过数据的工作结点上重新计算，还是直接从从外部文件系统读取数据。

●Driver（驱动结点）失效：如果运行Spark Streaming应用时驱动结点出现故障，那么很明显的StreamingContext已经丢失，同时在内存中的数据也全部丢失。对于这种情况，Spark Streaming应用程序在计算上有一个内在的结构——在每段micro-batch数据周期性地执行同样的Spark计算。这种结构允许把应用的状态（亦称checkpoint）周期性地保存到可靠的存储空间中，并在driver重新启动时恢复该状态。具体做法是在ssc.checkpoint（＜checkpoint directory＞）函数中进行设置，Spark Streaming就会定期把DStream的元信息写入到HDFS中，一旦驱动结点失效，丢失的StreamingContext会通过已经保存的检查点信息进行恢复。

最后我们谈一下Spark Stream的容错在Spark 1.2版本的一些改进：

实时流处理系统必须要能在7×24小时的时间内工作，因此它需要具备从各种系统故障中恢复过来的能力。最开始，SparkStreaming就支持从Driver和Worker故障恢复的能力。然而有些数据源的输入可能在故障恢复以后丢失数据。在Spark1.2版本中，Spark已经在SparkStreaming中对预写日志（也被称为journaling）作了初步支持，改进了恢复机制，并使更多数据源的零数据丢失有了可靠。

对于文件这样的源数据，Driver恢复机制足以做到零数据丢失，因为所有的数据都保存在了像HDFS或S3这样的容错文件系统中了。但对于像Kafka和Flume等其他数据源，有些接收到的数据还只缓存在内存中，尚未被处理，它们就有可能会丢失。这是由于Spark应用的分布式操作方式引起的。当Driver进程失败时，所有在standalone/yarn/mesos集群运行的Executor，连同它们在内存中的所有数据，也同时被终止。对于Spark Streaming来说，从诸如Kafka和Flume的数据源接收到的所有数据，在它们处理完成之前，一直都缓存在Execu-tor的内存中。纵然driver重新启动，这些缓存的数据也不能被恢复。为了避免这种数据损失，在Spark1.2发布版本中引进了预写日志（WriteAheadLogs）功能。(www.xing528.com)

预写日志功能的流程是：

1）一个SparkStreaming应用开始时（也就是Driver开始时），相关的StreamingContext使用SparkContext启动接收器成为长驻运行任务。这些接收器接收并保存流数据到Spark内存中以供处理。

2）接收器通知Driver。

3）接收块中的元数据（metadata）被发送到ApplicationDriver的StreamingContext。这个元数据包括：定位其在Executor内存中数据的块referenceid，块数据在日志中的偏移信息（如果启用了）。

用户传送数据的生命周期如图7-7所示。

图7-7 用户传送数据的生命周期

类似Kafka这样的系统可以通过复制数据保持可靠性。允许预写日志两次高效地复制同样的数据：一次由Kafka，而另一次由SparkStreaming。Spark未来版本将包含Kafka容错机制的原生支持，从而避免第二个日志。

2.DStream的持久化

与RDD一样，DStream同样也能通过persist（）方法将数据流存放在内存中，默认的持久化方式是MEMORY_ONLY_SER，也就是在内存中存放数据同时序列化的方式，这样做的好处是遇到需要多次迭代计算的程序时，速度优势十分的明显。而对于一些基于窗口的操作，如reduceByWindow、reduceByKeyAndWindow，以及基于状态的操作，如updateStateBykey，其默认的持久化策略就是保存在内存中。

对于来自网络的数据源（Kafka、Flume、sockets等），默认的持久化策略是将数据保存在两台机器上，这也是为了容错性而设计的。

另外，对于窗口和有状态的操作必须checkpoint，通过StreamingContext的checkpoint来指定目录，通过Dtream的checkpoint指定间隔时间，间隔必须是滑动间隔（slide interval）的倍数。