spark容错怎么实现

2019-02-26 17:19:05 hebaojing 阅读数 267

1. Executor容错
Spark支持多种运行模式,这些运行模型中的集群管理器会为任务分配运行资源,在运行资源中启动Executor,由Executor执行任务的运行,最终把任务运行状态发送给Driver。

在这里插入图片描述

(1)首先看Exeucutord的启动过程:在集群中由Master给应用程序分配运行资源后,然后再Worker中启动ExecutorRunner,而ExecutorRunner根据当前的运行模式启动CoarseGrainedExecutorBackend进程,当该进程会向Driver发送注册Executor信息,如果注册成功,则CoarseGrainedExecutorBackend在其内部启动Executor。Executor由ExecutorRunner进行管理,当Executor出现异常的时候(如所运行容器CoarseGrainedExecutorBackend进程异常退出等)由ExecutorRunner捕获当前异常并且发送ExecutorStateChanged消息给Worker进程。
(2)Worker进程接受到ExecutorStateChanged消息时,在Worker的handleExecutorStateChanged方法中,根据Executor状态消息更新,同时把Executor状态发送给Master。
(3)Master接收到Executor状态变化消息后,如果发现Executor出现异常退出,则调用Master.schedule方法,尝试获取可用的Worker节点并启动Executor,而这个Worker很可能不是失败之前运行的Executor的Worker节点。该尝试系统会进行10次,如果超过10次,则标记该应用运行失败并移除集群中该应用。

2. Worker异常
Spark独立运行模式采用的是Master/Slave的结构,其中Slave是由Worker来担任的,在运行的时候会发送心跳信息给Master,让Master知道Worker的实时状态,另一方面也会检测注册的Worker是否超时,因为在集群运行过程中,可能由于机器宕机或者进程被杀死等原因造成Worker进程异常退出。当Worker出现超时时,Master调用timeOutDeadWorker方法进行处理,在处理时根据Worker运行的是Executor和Driver进行分别处理。

  • 如果是Executor,Master先把该Worker上运行的Executor发送消息ExecutorUpdate给对应的Driver,告知Exeucotr已经丢失,同时把这些Executor从其应用程序中删除。参照前一节。
  • 如果是Drvier,则判断是否要重新启动。如果需要,则调用Master.schedule方法进行调度,分配合适节点重启Driver。

在这里插入图片描述

3. Master异常
Master作为Spark独立运行模式的核心,如果Master出现异常,则整个集群的运行资源将无法进行管理,整个集群将处于“群龙无首”的状态。很幸运的是,Spark在设计的时候考虑这个情况,在集群运行的时候,Master将启动一个或者多个StandBy Mster,当Master异常的时候,StandBy Mster将根据一定的规则确定其中一个为Master接管。

xxx

 

2019-08-19 11:17:59 qq_45518387 阅读数 315

Spark容错机制?

Spark的容错机制分为两种:
*第一种方式:*LineAge内部如果有分片数据丢失,可以通过上下依赖关系,启动重算机制,进行容错。
第二种方式:spark可以通过checkPoint机制,设置一个第三方存储路径(一般是hdfs),当有数据丢失的时候,从对应的路径恢复数据,但是一旦启动了checkPoint意味着之前的LineAge被终止。

2018-12-21 17:00:41 u012137473 阅读数 363

Spark容错机制,Lineage,Checkpoint

一般来说,分布式数据集的容错性有两种方式:数据检查点和记录数据的更新
面向大树据处理检查点机制的代价更高,需要通过数据中心的网络连接在不同的机器之间复制数据,而网络的带宽往往比内存的带宽低的多,并且需要消耗大量的存储资源。因此spark选择了记录数据的更新,但是记录的太细也会消耗大量的资源。因此,RDD只支持粗粒度的转换,即只记录单个块上的单个操作,然后将创建RDD的一系列变换序列(每个RDD都包含了他是如何由其他RDD变换过来的以及如何重建某一块数据的信息。因此RDD的容错机制又称“血统(Lineage)”)记录下来,以便恢复丢失的分区。
Lineage本质上很类似于数据库中的重做日志(Redo Log),只不过这个重做日志粒度很大,是对全局数据做同样的重做进而恢复数据。

Lineage机制

相比较其它系统的细粒度的内存数据更新级别的备份或是LOG机制,RDD的Lineage记录的是粗颗粒度的特定数据Transformation操作(filter,map,join等)当这个RDD的部分分区数据丢失,可以通过Lineage获取信息来重新计算和恢复丢失的数据。因为这种粗粒度的容错机制也限制了spark的运用场合,但相比细粒度的容错机制也大大提高了运行的性能。

RDD在Lineage依赖方面分为两种:窄依赖(Narrow Dependencies)与宽依赖(Wide Dependencies,源码中称为Shuffle Dependencies),用来解决数据容错的高效性。对于区分两种依赖可参考文章Spark stage划分和宽窄依赖

窄依赖和宽依赖的概念主要用在两个地方:一个是容错中相当于Redo日志的功能;另一个是在调度中构建DAG作为不同Stage的划分点。

对于宽依赖,Stage计算的输入和输出在不同的节点上,lineage方法对与输入节点完好,而输出节点宕机时,通过重新计算,这种情况下,这种方法容错是有效的,否则无效,因为无法重试,需要向上其祖先追溯看是否可以重试(这就是lineage,血统的意思),Narrow Dependencies对于数据的重算开销要远小于Wide Dependencies的数据重算开销。

容错原理

在容错机制中,如果一个节点死机了,而且运算窄依赖,则只要把丢失的父RDD分区重算即可,不依赖于其他节点。而宽依赖需要父RDD的所有分区都存在,重算就很昂贵了。可以这样理解开销的经济与否:在窄依赖中,在子RDD的分区丢失、重算父RDD分区时,父RDD相应分区的所有数据都是子RDD分区的数据,并不存在冗余计算。在宽依赖情况下,丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的,会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据,这样就会产生冗余计算开销,这也是宽依赖开销更大的原因。因此如果使用Checkpoint算子来做检查点,不仅要考虑Lineage是否足够长,也要考虑是否有宽依赖,对宽依赖加Checkpoint是最物有所值的。

Checkpoint机制

通过上述分析可以看出在以下两种情况下,RDD需要加检查点。

  1. DAG中的Lineage过长,如果重算,则开销太大(如在PageRank中)。
  2. 在宽依赖上做Checkpoint获得的收益更大。

由于RDD是只读的,所以Spark中计算中数据一致性不是主要关心的,内存相对容易管理,这也是作者牛B的地方,减少了框架的复杂性,提升了性能和可扩展性,为上层框架的提供了强有力的基础。在RDD计算中通过Checkpoint机制进行容错,对于传统Checkpoint一般有两种方式,冗余数据和日志记录更新操作。在RDD的doCheckpoint相当于冗余数据来缓存数据,而上面介绍的Lineage相当于粗粒度的记录更新操作来实现容错。

检查点(本质是将RDD写入磁盘做检查点) 是为了Lineage做容做的辅助,Lineage过长会导致容做成本过高,可以在中间做检查点,如果之后有节点出问题而丢失数据,从检查点开始做Lineage,减少开销。

下面是RDD源码中Checkpoint的方法,里面建议在执行Checkpoint()方法之前先对rdd进行persisted操作,并且需要在SparkContext中设置Checkpoint的路径。

  /**
   * Mark this RDD for Checkpointing. It will be saved to a file inside the Checkpoint
   * directory set with `SparkContext#setCheckpointDir` and all references to its parent
   * RDDs will be removed. This function must be called before any job has been
   * executed on this RDD. It is strongly recommended that this RDD is persisted in
   * memory, otherwise saving it on a file will require recomputation.
   */
   def Checkpoint(): Unit = RDDCheckpointData.synchronized {
    // NOTE: we use a global lock here due to complexities downstream with ensuring
    // children RDD partitions point to the correct parent partitions. In the future
    // we should revisit this consideration.
    if (context.CheckpointDir.isEmpty) {
      throw new SparkException("Checkpoint directory has not been set in the SparkContext")
    } else if (CheckpointData.isEmpty) {
      CheckpointData = Some(new ReliableRDDCheckpointData(this))
    }
  }

Checkpoint 写流程

checkpoint这个函数调用中,所依赖的RDD都会被删除,函数必须在job运行之前调用执行,强烈建议RDD缓存在内存中,否则保存到文件的时候需要从头计算。初始化RDD的CheckpointData变量为ReliableRDDCheckpointData。这时候标记为Initialized状态。

在所有job action的时候,runJob方法中都会调用rdd.doCheckpoint,这个会向前递归调用所有的依赖的RDD,看看需不需要Checkpoint。需要需要Checkpoint,然后调用CheckpointData.get.Checkpoint(),里面标记状态为CheckpointingInProgress,里面调用具体实现类的ReliableRDDCheckpointData的doCheckpoint方法。

doCheckpoint->writeRDDToCheckpointDirectory,注意这里会把job再运行一次,如果已经cache了,就可以直接使用缓存中的RDD了,就不需要重头计算一遍了,这时候直接把RDD,输出到hdfs,每个分区一个文件,会先写到一个临时文件,如果全部输出完,进行rename,如果输出失败,就回滚delete。

标记状态为Checkpointed,markCheckpointed方法中清除所有的依赖,怎么清除依赖的呢,就是把RDD变量的强引用设置为null,垃圾回收了,会触发ContextCleaner里面监听清除实际BlockManager缓存中的数据。

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,来处理依赖和获取分区。
如果没有,才向前回溯依赖。

参考文章:
https://blog.csdn.net/u012137473/article/details/85007483
https://www.cnblogs.com/duanxz/p/6329675.html
https://blog.csdn.net/wt346326775/article/details/72870518?utm_source=blogxgwz6

2017-06-23 10:57:12 dengxing1234 阅读数 26610

容错方式

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

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

1、Lineage机制

Lineage简介

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

两种依赖关系

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分区来区分窄依赖(父分区对应一个子分区)和宽依赖(父分区对应多个子分
区)。如果对应多个,则当容错重算分区时,因为父分区数据只有一部分是需要重算子分区的,其余数据重算就造成了冗余计算。

对于宽依赖,Stage计算的输入和输出在不同的节点上,对于输入节点完好,而输出节点死机的情况,通过重新计算恢复数据这种情况下,这种方法容错是有效的,否则无效,因为无法重试,需要向上追溯其祖先看是否可以重试(这就是lineage,血统的意思),窄依赖对于数据的重算开销要远小于宽依赖的数据重算开销。

窄依赖和宽依赖的概念主要用在两个地方:一个是容错中相当于Redo日志的功能;另一个是在调度中构建DAG作为不同Stage的划分点。

依赖关系的特性

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

容错原理

在容错机制中,如果一个节点死机了,而且运算窄依赖,则只要把丢失的父RDD分区重算即可,不依赖于其他节点。而宽依赖需要父RDD的所有分区都存在,重算就很昂贵了。可以这样理解开销的经济与否:在窄依赖中,在子RDD的分区丢失、重算父RDD分区时,父RDD相应分区的所有数据都是子RDD分区的数据,并不存在冗余计算。在宽依赖情况下,丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的,会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据,这样就会产生冗余计算开销,这也是宽依赖开销更大的原因。

2、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,就会减少开销。

2015-07-14 19:48:22 JasonDing1354 阅读数 4175

引入

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

Lineage机制

Lineage简介

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

两种依赖关系

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)。

本质理解:根据父RDD分区是对应1个还是多个子RDD分区来区分窄依赖(父分区对应一个子分区)和宽依赖(父分区对应多个子分
区)。如果对应多个,则当容错重算分区时,因为父分区数据只有一部分是需要重算子分区的,其余数据重算就造成了冗余计算。

对于宽依赖,Stage计算的输入和输出在不同的节点上,对于输入节点完好,而输出节点死机的情况,通过重新计算恢复数据这种情况下,这种方法容错是有效的,否则无效,因为无法重试,需要向上追溯其祖先看是否可以重试(这就是lineage,血统的意思),窄依赖对于数据的重算开销要远小于宽依赖的数据重算开销。

窄依赖和宽依赖的概念主要用在两个地方:一个是容错中相当于Redo日志的功能;另一个是在调度中构建DAG作为不同Stage的划分点。

依赖关系的特性

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

容错原理

在容错机制中,如果一个节点死机了,而且运算窄依赖,则只要把丢失的父RDD分区重算即可,不依赖于其他节点。而宽依赖需要父RDD的所有分区都存在,重算就很昂贵了。可以这样理解开销的经济与否:在窄依赖中,在子RDD的分区丢失、重算父RDD分区时,父RDD相应分区的所有数据都是子RDD分区的数据,并不存在冗余计算。在宽依赖情况下,丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的,会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据,这样就会产生冗余计算开销,这也是宽依赖开销更大的原因。因此如果使用Checkpoint算子来做检查点,不仅要考虑Lineage是否足够长,也要考虑是否有宽依赖,对宽依赖加Checkpoint是最物有所值的。

Checkpoint机制

通过上述分析可以看出在以下两种情况下,RDD需要加检查点。

  1. DAG中的Lineage过长,如果重算,则开销太大(如在PageRank中)。
  2. 在宽依赖上做Checkpoint获得的收益更大。

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

检查点(本质是通过将RDD写入Disk做检查点)是为了通过lineage做容错的辅助,lineage过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果之后有节点出现问题而丢失分区,从做检查点的RDD开始重做Lineage,就会减少开销。

转载请注明作者Jason Ding及其出处
GitCafe博客主页(http://jasonding1354.gitcafe.io/)
Github博客主页(http://jasonding1354.github.io/)
CSDN博客(http://blog.csdn.net/jasonding1354)
简书主页(http://www.jianshu.com/users/2bd9b48f6ea8/latest_articles)
Google搜索jasonding1354进入我的博客主页

SparkRDD容错处理

阅读数 1090

spark容错机制

阅读数 332