Spark Streaming 数据清理机制

雨点打透心脏的1/2处 2022-07-05 15:08 217阅读 0赞

# Spark Streaming 数据清理机制 #

时间 2016-05-11 10:29:54   [51CTO][]

原文   [http://developer.51cto.com/art/201605/511024.htm][http_developer.51cto.com_art_201605_511024.htm]

主题  [Spark Streaming][]

## Spark Streaming 数据清理机制 ##

大家刚开始用Spark Streaming时，心里肯定嘀咕，对于一个7\*24小时运行的数据，cache住的RDD,broadcast 系统会帮忙自己清理掉么？还a是说必须自己做清理？如果系统帮忙清理的话，机制是啥？

作者：祝威廉 来源：简书 |  *2016-05-11 10:29*

## [51CTO网+ 首届中国APP创新评选大赛>>][51CTO_ _APP] ##

#### 前言 ####

为啥要了解机制呢?这就好比JVM的垃圾回收，虽然JVM的垃圾回收已经巨牛了，但是依然会遇到很多和它相关的case导致系统运行不正常。

这个内容我记得自己刚接触Spark Streaming的时候，老板也问过我，运行期间会保留多少个RDD? 当时没回答出来。后面在群里也有人问到了，所以就整理了下。文中如有谬误之处，还望指出。

#### DStream 和 RDD ####

我们知道Spark Streaming 计算还是基于Spark Core的，Spark Core 的核心又是RDD. 所以Spark Streaming 肯定也要和RDD扯上关系。然而Spark Streaming 并没有直接让用户使用RDD而是自己抽象了一套DStream的概念。 DStream 和 RDD 是包含的关系，你可以理解为Java里的装饰模式，也就是DStream 是对RDD的增强，但是行为表现和RDD是基本上差不多的。都具备几个条件：

具有类似的tranformation动作，比如map,reduceByKey等，也有一些自己独有的，比如Window，mapWithStated等

都具有Action动作，比如foreachRDD，count等

从编程模型上看是一致的。

所以很可能你写的那堆Spark Streaming代码看起来好像和Spark 一致的,然而并不能直接复用，因为一个是DStream的变换，一个是RDD的变化。

#### Spark Streaming中 DStream 介绍 ####

DStream 下面包含几个类：

*  数据源类，比如InputDStream,具体如DirectKafkaInputStream等
 *  转换类，典型比如MappedDStream,ShuffledDStream
 *  输出类，典型比如ForEachDStream

从上面来看，数据从开始(输入)到结束(输出)都是DStream体系来完成的，也就意味着用户正常情况是无法直接去产生和操作RDD的，这也就是说，DStream有机会和义务去负责RDD的生命周期。

这就回答了前言中的问题了。Spark Streaming具备自动清理功能。

#### RDD 在Spark Stream中产生的流程 ####

在Spark Streaming中RDD的生命流程大体如下：

*  在InputDStream会将接受到的数据转化成RDD,比如DirectKafkaInputStream 产生的就是 KafkaRDD
 *  接着通过MappedDStream等进行数据转换，这个时候是直接调用RDD对应的map方法进行转换的
 *  在进行输出类操作时，才暴露出RDD,可以让用户执行相应的存储，其他计算等操作。

我们这里就以下面的代码来进行更详细的解释：

val source  =   KafkaUtils.createDirectInputStream(....) 
    source.map(....).foreachRDD{rdd=> 
        rdd.saveTextFile(....) 
    }

foreachRDD 产生ForEachDStream，因为foreachRDD是个Action,所以会触发任务的执行，会被调用generateJob方法。

override def generateJob(time: Time): Option[Job] = { 
       parent.getOrCompute(time) match { 
         case Some(rdd) => 
           val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) { 
             foreachFunc(rdd, time) 
           } 
           Some(new Job(time, jobFunc)) 
         case None => None 
       } 
     }

对应的parent是MappedDStream,也就是说调用MappedDStream.getOrCompute.该方法在DStream中，首先会在MappedDStream对象中的generatedRDDs 变量中查找是否已经有RDD,如果没有则触发计算，并且将产生的RDD放到generatedRDDs

@transientprivate[streaming] var generatedRDDs = new HashMap[Time, RDD[T]] () 
     
    private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = { 
        // If RDD was already generated, then retrieve it from HashMap, 
        // or else compute the RDD 
        generatedRDDs.get(time).orElse { 
    .... 
    generatedRDDs.put(time, newRDD) 
    ....

计算RDD是调用的compute方法，MappedDStream 的compute方法很简单，直接调用的父类也就是DirectKafkaInputStream的getOrCompute方法：

override def compute(validTime: Time): Option[RDD[U]] = { 
        parent.getOrCompute(validTime).map(_.map[U](mapFunc)) 
      }

在上面的例子中，MappedDStream 的parent是DirectKafkaInputStream中,这是个数据源，所以他的compute方法会直接new出一个RDD.

从上面可以得出几个结论：

*  数据源以及转换类DStream都会维护一个generatedRDDs，可以按batchTime 进行获取
 *  内部本质还是进行的RDD的转换
 *  如果我们调用了cache会发生什么

这里又会有两种情况，一种是调用DStream.cache,第二种是RDD.cache。事实上他们是完全一样的。

1.  DStream的cache 动作只是将DStream的变量storageLevel 设置为MEMORY\_ONLY\_SER，然后在产生(或者获取)RDD的时候，调用RDD的persit方法进行设置。所以DStream.cache 产生的效果等价于RDD.cache(也就是你自己调用foreachRDD 将RDD 都设置一遍)
2.  进入正题，我们是怎么释放Cache住的RDD的

其实无所谓Cache不Cache住，RDD最终都是要释放的，否则运行久了，光RDD对象也能承包了你的内存。我们知道，在Spark Streaming中，周期性产生事件驱动Spark Streaming 的类其实是:

org.apache.spark.streaming.scheduler.JobGenerator

他内部有个永动机(定时器),定时发布一个产生任务的事件:

private val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds, longTime => eventLoop.post(GenerateJobs(new Time(longTime))), "JobGenerator")

然后通过processEvent进行事件处理：

/** Processes all events */ 
     private def processEvent(event: JobGeneratorEvent) { 
       logDebug("Got event " + event) 
       event match { 
         case GenerateJobs(time) => generateJobs(time) 
         case ClearMetadata(time) => clearMetadata(time) 
         case DoCheckpoint(time, clearCheckpointDataLater) => 
           doCheckpoint(time, clearCheckpointDataLater) 
         case ClearCheckpointData(time) => clearCheckpointData(time) 
       } 
     }

目前我们只关注ClearMetadata 事件。对应的方法为：

private def clearMetadata(time: Time) { 
        ssc.graph.clearMetadata(time) 
     
        // If checkpointing is enabled, then checkpoint, 
        // else mark batch to be fully processed 
        if (shouldCheckpoint) { 
          eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = true)) 
        } else { 
          // If checkpointing is not enabled, then delete metadata information about 
          // received blocks (block data not saved in any case). Otherwise, wait for 
          // checkpointing of this batch to complete. 
          val maxRememberDuration = graph.getMaxInputStreamRememberDuration() 
          jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration) 
          jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration) 
          markBatchFullyProcessed(time) 
        } 
      }

首先是清理输出DStream(比如ForeachDStream),接着是清理输入类(基于Receiver模式)的数据。

ForeachDStream 其实调用的也是DStream的方法。该方法大体如下：

private[streaming] def clearMetadata(time: Time) { 
        val unpersistData = ssc.conf.getBoolean("spark.streaming.unpersist", true) 
        val oldRDDs = generatedRDDs.filter(_._1 <= (time - rememberDuration)) 
        logDebug("Clearing references to old RDDs: [" + 
          oldRDDs.map(x => s"${x._1} -> ${x._2.id}").mkString(", ") + "]") 
        generatedRDDs --= oldRDDs.keys 
        if (unpersistData) { 
          logDebug("Unpersisting old RDDs: " + oldRDDs.values.map(_.id).mkString(", ")) 
          oldRDDs.values.foreach { rdd => 
            rdd.unpersist(false) 
            // Explicitly remove blocks of BlockRDD 
            rdd match { 
              case b: BlockRDD[_] => 
                logInfo("Removing blocks of RDD " + b + " of time " + time) 
                b.removeBlocks() 
              case _ => 
            } 
          } 
        } 
        logDebug("Cleared " + oldRDDs.size + " RDDs that were older than " + 
          (time - rememberDuration) + ": " + oldRDDs.keys.mkString(", ")) 
        dependencies.foreach(_.clearMetadata(time)) 
      }

大体执行动作如下描述：

1.  根据记忆周期得到应该剔除的RDD
2.  根据是否要清理cache数据，进行unpersit 操作，并且显示的移除block
3.  根据依赖调用其他的DStream进行动作清理

这里我们还可以看到，通过参数spark.streaming.unpersist 你是可以决定是否手工控制是否需要对cache住的数据进行清理。

这里你会有两个疑问：

1.  dependencies 是什么?
2.  rememberDuration 是怎么来的?

dependencies 你可以简单理解为父DStream,通过dependencies 我们可以获得已完整DStream链。

rememberDuration 的设置略微复杂些,大体是 slideDuration,如果设置了checkpointDuration 则是2\*checkpointDuration 或者通过DStreamGraph.rememberDuration(如果设置了的话，譬如通过StreamingContext.remember方法,不过通过该方法设置的值要大于计算得到的值会生效)

另外值得一提的就是后面的DStream 会调整前面的DStream的rememberDuration，譬如如果你用了window\* 相关的操作，则在此之前的DStream 的rememberDuration 都需要加上windowDuration。

然后根据Spark Streaming的定时性，每个周期只要完成了，都会触发清理动作,这个就是清理动作发生的时机。代码如下：

def onBatchCompletion(time: Time) {      
        eventLoop.post(ClearMetadata(time)) 
    }

#### 总结下 ####

Spark Streaming 会在每个Batch任务结束时进行一次清理动作。每个DStream 都会被扫描，不同的DStream根据情况不同，保留的RDD数量也是不一致的，但都是根据rememberDuration变量决定,而该变量会被下游的DStream所影响，所以不同的DStream的rememberDuration取值是不一样的。

【编辑推荐】

1.  [Spark Streaming 的玫瑰与刺][Spark Streaming 1]
2.  [Spark Streaming 妙用之实现工作流调度器][Spark Streaming 2]
3.  [借助 Redis ，让 Spark 提速 45 倍！][Redis _ Spark _ 45]
4.  [LinkedIn开源Dr. Elephant:Hadoop和Spark性能监控工具][LinkedIn_Dr. Elephant_Hadoop_Spark]
5.  [Spark与Storm注意：Apache Apex已然横空出世][Spark_Storm_Apache Apex]

[51CTO]: http://www.tuicool.com/sites/rMjaeyJ
[http_developer.51cto.com_art_201605_511024.htm]: http://developer.51cto.com/art/201605/511024.htm?utm_source=tuicool&utm_medium=referral
[Spark Streaming]: http://www.tuicool.com/topics/11090153
[51CTO_ _APP]: http://mdsa.51cto.com/act/2015/applications/
[Spark Streaming 1]: http://developer.51cto.com/art/201601/504887.htm
[Spark Streaming 2]: http://developer.51cto.com/art/201603/506796.htm
[Redis _ Spark _ 45]: http://developer.51cto.com/art/201603/507663.htm
[LinkedIn_Dr. Elephant_Hadoop_Spark]: http://developer.51cto.com/art/201604/509314.htm
[Spark_Storm_Apache Apex]: http://developer.51cto.com/art/201604/510227.htm