【总结】Spark容错机制

时间  2020-05-16 标签 总结 spark 容错 机制

容错方式

容错 指的是一个系统在部分模块出现故障时还可否持续的对外提供服务，一个高可用的系统应该具备很高的容错性；对于一个大的集群系统来讲，机器故障、网络异常等都是很常见的，Spark这样的大型分布式计算集群提供了不少的容错机制来提升整个系统的可用性。
node

通常来讲，分布式数据集的容错性有两种方式：数据检查点和记录数据的更新。
面向大规模数据分析，数据检查点操做成本很高，须要经过数据中心的网络链接在机器之间复制庞大的数据集，而网络带宽每每比内存带宽低得多，同时还须要消耗更多的存储资源。
所以，Spark选择记录更新的方式。可是，若是更新粒度太细太多，那么记录更新成本也不低。所以，RDD只支持粗粒度转换，即只记录单个块上执行的单个操做，而后将建立RDD的一系列变换序列（每一个RDD都包含了他是如何由其余RDD变换过来的以及如何重建某一块数据的信息。所以RDD的容错机制又称“血统(Lineage)”容错）记录下来，以便恢复丢失的分区。
Lineage本质上很相似于数据库中的重作日志（Redo Log），只不过这个重作日志粒度很大，是对全局数据作一样的重作进而恢复数据。数据库

一、Lineage机制

Lineage简介

相比其余系统的细颗粒度的内存数据更新级别的备份或者LOG机制，RDD的Lineage记录的是粗颗粒度的特定数据Transformation操做（如filter、map、join等）行为。当这个RDD的部分分区数据丢失时，它能够经过Lineage获取足够的信息来从新运算和恢复丢失的数据分区。由于这种粗颗粒的数据模型，限制了Spark的运用场合，因此Spark并不适用于全部高性能要求的场景，但同时相比细颗粒度的数据模型，也带来了性能的提高。apache

两种依赖关系

RDD在Lineage依赖方面分为两种：窄依赖(Narrow Dependencies)与宽依赖(Wide Dependencies,源码中称为Shuffle
Dependencies)，用来解决数据容错的高效性。 缓存

  • 窄依赖是指父RDD的每个分区最多被一个子RDD的分区所用，表现为一个父RDD的分区对应于一个子RDD的分区
    或多个父RDD的分区对应于一个子RDD的分区，也就是说一个父RDD的一个分区不可能对应一个子RDD的多个分区。
    1个父RDD分区对应1个子RDD分区，这其中又分两种状况：1个子RDD分区对应1个父RDD分区（如map、filter等算子），1个子RDD分区对应N个父RDD分区（如co-paritioned（协同划分）过的Join）。
  • 宽依赖是指子RDD的分区依赖于父RDD的多个分区或全部分区，即存在一个父RDD的一个分区对应一个子RDD的多个分区。
    1个父RDD分区对应多个子RDD分区，这其中又分两种状况：1个父RDD对应全部子RDD分区（未经协同划分的Join）或者1个父RDD对应非所有的多个RDD分区（如groupByKey）。

 spark 依赖的实现：网络

abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
    //返回子RDD的partitionId依赖的全部的parent RDD的Partition(s)
    def getParents(partitionId: Int): Seq[Int]
    override def rdd: RDD[T] = _rdd
}

（1）窄依赖是有两种具体实现，分别以下：

     一种是一对一的依赖，即OneToOneDependency：框架

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
    override def getParents(partitionId: Int) = List(partitionId)
}

    经过getParents的实现不难看出，RDD仅仅依赖于parent RDD相同ID的Partition。
 
 
    还有一个是范围的依赖，即RangeDependency，它仅仅被org.apache.spark.rdd.UnionRDD使用。UnionRDD是把多个RDD合成一个RDD，这些RDD是被拼接而成，即每一个parent RDD的Partition的相对顺序不会变，只不过每一个parent RDD在UnionRDD中的Partition的起始位置不一样。所以它的getPartents以下： 
override def getParents(partitionId: Int) = {
    if(partitionId >= outStart && partitionId < outStart + length) {
       List(partitionId - outStart + inStart)
    } else {
       Nil
    }
}

　　其中，inStart是parent RDD中Partition的起始位置，outStart是在UnionRDD中的起始位置，length就是parent RDD中Partition的数量。

（2）宽依赖的实现

　　宽依赖的实现只有一种：ShuffleDependency。子RDD依赖于parent RDD的全部Partition，所以须要Shuffle过程：分布式

class ShuffleDependency[K, V, C](
    @transient _rdd: RDD[_ <: Product2[K, V]],
    val partitioner: Partitioner,
    val serializer: Option[Serializer] = None,
    val keyOrdering: Option[Ordering[K]] = None,
    val aggregator: Option[Aggregator[K, V, C]] = None,
    val mapSideCombine: Boolean = false)
extends Dependency[Product2[K, V]] {
 
override def rdd = _rdd.asInstanceOf[RDD[Product2[K, V]]]
//获取新的shuffleId
val shuffleId: Int = _rdd.context.newShuffleId()
//向ShuffleManager注册Shuffle的信息
val shuffleHandle: ShuffleHandle =
_rdd.context.env.shuffleManager.registerShuffle(
    shuffleId, _rdd.partitions.size, this)
 
    _rdd.sparkContext.cleaner.foreach(_.registerShuffleForCleanup(this))
}
　　注意：宽依赖支持两种Shuffle Manager。

　　即org.apache.spark.shuffle.hash.HashShuffleManager（基于Hash的Shuffle机制）和org.apache.spark.shuffle.sort.SortShuffleManager（基于排序的Shuffle机制）。
 

本质理解：根据父RDD分区是对应1个仍是多个子RDD分区来区分窄依赖（父分区对应一个子分区）和宽依赖（父分区对应多个子分
区）。若是对应多个，则当容错重算分区时，由于父分区数据只有一部分是须要重算子分区的，其他数据重算就形成了冗余计算。ide

对于宽依赖，Stage计算的输入和输出在不一样的节点上，对于输入节点无缺，而输出节点死机的状况，经过从新计算恢复数据这种状况下，这种方法容错是有效的，不然无效，由于没法重试，须要向上追溯其祖先看是否能够重试（这就是lineage，血统的意思），窄依赖对于数据的重算开销要远小于宽依赖的数据重算开销。性能

窄依赖和宽依赖的概念主要用在两个地方：一个是容错中至关于Redo日志的功能；另外一个是在调度中构建DAG做为不一样Stage的划分点。this

依赖关系的特性

第一，窄依赖能够在某个计算节点上直接经过计算父RDD的某块数据计算获得子RDD对应的某块数据；宽依赖则要等到父RDD全部数据都计算完成以后，而且父RDD的计算结果进行hash并传到对应节点上以后才能计算子RDD。
第二，数据丢失时，对于窄依赖只须要从新计算丢失的那一块数据来恢复；对于宽依赖则要将祖先RDD中的全部数据块所有从新计算来恢复。因此在长“血统”链特别是有宽依赖的时候，须要在适当的时机设置数据检查点。也是这两个特性要求对于不一样依赖关系要采起不一样的任务调度机制和容错恢复机制。

容错原理

在容错机制中，若是一个节点死机了，并且运算窄依赖，则只要把丢失的父RDD分区重算便可，不依赖于其余节点。而宽依赖须要父RDD的全部分区都存在，重算就很昂贵了。能够这样理解开销的经济与否：在窄依赖中，在子RDD的分区丢失、重算父RDD分区时，父RDD相应分区的全部数据都是子RDD分区的数据，并不存在冗余计算。在宽依赖状况下，丢失一个子RDD分区重算的每一个父RDD的每一个分区的全部数据并非都给丢失的子RDD分区用的，会有一部分数据至关于对应的是未丢失的子RDD分区中须要的数据，这样就会产生冗余计算开销，这也是宽依赖开销更大的缘由。

二、Checkpoint机制

咱们应该都很熟悉 checkpoint 这个概念， 就是把内存中的变化刷新到持久存储，斩断依赖链 在存储中 checkpoint 是一个很常见的概念， 举几个例子

  • 数据库 checkpoint 过程当中通常把内存中的变化进行持久化到物理页， 这时候就能够斩断依赖链， 就能够把 redo 日志删掉了， 而后更新下检查点，
  • hdfs namenode 的元数据 editlog， Secondary namenode 会把 edit log 应用到 fsimage， 而后刷到磁盘上， 也至关于作了一次 checkpoint， 就能够把老的 edit log 删除了。
  • spark streaming 中对于一些 有状态的操做， 这在某些 stateful 转换中是须要的，在这种转换中，生成 RDD 须要依赖前面的 batches，会致使依赖链随着时间而变长。为了不这种没有尽头的变长，要按期将中间生成的 RDDs 保存到可靠存储来切断依赖链， 必须隔一段时间进行一次进行一次 checkpoint。
cache 和 checkpoint 是有显著区别的， 缓存把 RDD 计算出来而后放在内存中， 可是RDD 的依赖链（至关于数据库中的redo 日志）， 也不能丢掉， 当某个点某个 executor 宕了， 上面cache 的RDD就会丢掉， 须要经过 依赖链重放计算出来， 不一样的是， checkpoint 是把 RDD 保存在 HDFS中， 是多副本可靠存储，因此依赖链就能够丢掉了，就斩断了依赖链， 是经过复制实现的高容错。可是有一点要注意， 由于checkpoint是须要把 job 从新从头算一遍， 最好先cache一下， checkpoint就能够直接保存缓存中的 RDD 了， 就不须要重头计算一遍了， 对性能有极大的提高。

checkpoint 的正确使用姿式 

val data = sc.textFile("/tmp/spark/1.data").cache() // 注意要cache 
sc.setCheckpointDir("/tmp/spark/checkpoint")
data.checkpoint 
data.count

使用很简单， 就是设置一下 checkpoint 目录，而后再rdd上调用 checkpoint 方法， action 的时候就对数据进行了 checkpoint

checkpoint 写流程

RDD checkpoint 过程当中会通过如下几个状态，

[ Initialized –> marked for checkpointing –> checkpointing in progress –> checkpointed ]

咱们看下状态转换流程

  •  首先 driver program 须要使用 rdd.checkpoint() 去设定哪些 rdd 须要 checkpoint，设定后，该 rdd 就接受 RDDCheckpointData 管理。用户还要设定 checkpoint 的存储路径，通常在 HDFS 上。
  • marked for checkpointing：初始化后，RDDCheckpointData 会将 rdd 标记为 MarkedForCheckpoint。
  • checkpointing in progress：每一个 job 运行结束后会调用 finalRdd.doCheckpoint()，finalRdd 会顺着 computing chain 回溯扫描，碰到要 checkpoint 的 RDD 就将其标记为 CheckpointingInProgress，而后将写磁盘（好比写 HDFS）须要的配置文件（如 core-site.xml 等）broadcast 到其余 worker 节点上的 blockManager。完成之后，启动一个 job 来完成 checkpoint（使用 rdd.context.runJob(rdd, CheckpointRDD.writeToFile(path.toString, broadcastedConf))）。
  • checkpointed：job 完成 checkpoint 后，将该 rdd 的 dependency 所有清掉，并设定该 rdd 状态为 checkpointed。而后，为该 rdd 强加一个依赖，设置该 rdd 的 parent rdd 为 CheckpointRDD，该 CheckpointRDD 负责之后读取在文件系统上的 checkpoint 文件，生成该 rdd 的 partition。

checkpoint 读流程

若是一个RDD 咱们已经 checkpoint了那么是何时用呢， checkpoint 将 RDD 持久化到 HDFS 或本地文件夹，若是不被手动 remove 掉，是一直存在的，也就是说能够被下一个 driver program 使用。 好比 spark streaming 挂掉了， 重启后就可使用以前 checkpoint 的数据进行 recover （这个流程咱们在下面一篇文章会讲到） ， 固然在同一个 driver program 也可使用。 咱们讲下在同一个 driver program 中是怎么使用 checkpoint 数据的。

若是 一个 RDD 被checkpoint了， 若是这个 RDD 上有 action 操做时候，或者回溯的这个 RDD 的时候,这个 RDD 进行计算的时候，里面判断若是已经 checkpoint 过, 对分区和依赖的处理都是使用的 RDD 内部的 checkpointRDD 变量。

具体细节以下，

若是 一个 RDD 被checkpoint了， 那么这个 RDD 中对分区和依赖的处理都是使用的 RDD 内部的 checkpointRDD 变量， 具体实现是 ReliableCheckpointRDD 类型。 这个是在 checkpoint 写流程中建立的。依赖和获取分区方法中先判断是否已经checkpoint， 若是已经checkpoint了， 就斩断依赖， 使用ReliableCheckpointRDD， 来处理依赖和获取分区。

若是没有，才往前回溯依赖。 依赖就是没有依赖， 由于已经斩断了依赖， 获取分区数据就是读取 checkpoint 到 hdfs目录中不一样分区保存下来的文件。

整个 checkpoint 读流程就完了。


在如下两种状况下，RDD须要加检查点。

  1. DAG中的Lineage过长，若是重算，则开销太大（如在PageRank中）。
  2. 在宽依赖上作Checkpoint得到的收益更大。

因为RDD是只读的，因此Spark的RDD计算中一致性不是主要关心的内容，内存相对容易管理，这也是设计者颇有远见的地方，这样减小了框架的复杂性，提高了性能和可扩展性，为之后上层框架的丰富奠基了强有力的基础。
在RDD计算中，经过检查点机制进行容错，传统作检查点有两种方式：经过冗余数据和日志记录更新操做。在RDD中的doCheckPoint方法至关于经过冗余数据来缓存数据，而以前介绍的血统就是经过至关粗粒度的记录更新操做来实现容错的。

检查点（本质是经过将RDD写入Disk作检查点）是为了经过lineage作容错的辅助，lineage过长会形成容错成本太高，这样就不如在中间阶段作检查点容错，若是以后有节点出现问题而丢失分区，从作检查点的RDD开始重作Lineage，就会减小开销。

联系我们 沪ICP备2021010478号-7