spark源码系列之累加器实现机制及自定义累加器

一,基本概念

累加器是Spark的一种变量,顾名思义该变量只能增加。有以下特点:

1,累加器只能在Driver端构建及并只能是Driver读取结果,Task只能累加。

2,累加器不会改变Spark Lazy计算的特点。只会在Job触发的时候进行相关累加操作。

3,现有累加器的类型。
相信有很多学习大数据的道友,在这里我给大家说说我滴群哦,大数据海量知识分享,784789432.在此我保证,绝对大数据的干货,等待各位的到来,我们一同从入门到精通吧!

二,累加器的使用

Driver端初始化,并在Action之后获取值。

val accum = sc.accumulator(0, "test Accumulator")
accum.value

Executor端进行计算

accum+=1;

三,累加器的重点类

Class Accumulator extends Accumulable

主要是实现了累加器的初始化及封装了相关的累加器操作方法。同时在类对象构建的时候向我们的Accumulators注册了累加器。累加器的add操作的返回值类型和我们传入的值类型可以不一样。所以,我们一定要定义好如何累加和合并值。也即add方法

object Accumulators:

该方法在Driver端管理着我们的累加器,也包含了特定累加器的聚合操作。

trait AccumulatorParam[T] extends AccumulableParam[T, T]:

AccumulatorParam的addAccumulator操作的泛型封装,具体的实现还是要再具体实现类里面实现addInPlace方法。

object AccumulatorParam:

主要是进行隐式类型转换的操作。

TaskContextImpl:

在Executor端管理着我们的累加器。

四,累加器的源码解析

1,Driver端的初始化

val accum = sc.accumulator(0, "test Accumulator")

val acc = new Accumulator(initialValue, param, Some(name))

主要是在Accumulable(Accumulator)中调用了,这样我们就可以使用Accumulator使用了。

Accumulators.register(this)

2,Executor端的反序列化得到我们对象的过程

首先,我们的value_ 可以看到其并不支持序列化

@volatile @transient private var value_ : R = initialValue // Current value on master

其初始化是在我们反序列化的时候做的,反序列化还完成了Accumulator向我们的TaskContextImpl的注册

反序列化是在调用ResultTask的RunTask方法的时候做的

val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)

过程中会调用

private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
in.defaultReadObject()
value_ = zero
deserialized = true
// Automatically register the accumulator when it is deserialized with the task closure.
//
// Note internal accumulators sent with task are deserialized before the TaskContext is created
// and are registered in the TaskContext constructor. Other internal accumulators, such SQL
// metrics, still need to register here.
val taskContext = TaskContext.get()
if (taskContext != null) {
taskContext.registerAccumulator(this)
}
}

3,累加器的累加

accum+=1;

param.addAccumulator(value_, term)

根据不同的累加器参数有不同的实现AccumulableParam

如,int类型。最终调用的AccumulatorParam特质的addAccumulator方法。

trait AccumulatorParam[T] extends AccumulableParam[T, T] {
def addAccumulator(t1: T, t2: T): T = {
addInPlace(t1, t2)
}
}

然后,调用的是各个具体实现的addInPlace方法

implicit object IntAccumulatorParam extends AccumulatorParam[Int] {
def addInPlace(t1: Int, t2: Int): Int = t1 + t2
def zero(initialValue: Int): Int = 0
}

返回后更新了我们的Accumulators的value_的值。

4,Accumulator的各个节点累加的之后的聚合操作

在Task类的run方法里面得到并返回的

(runTask(context), context.collectAccumulators())

最终在DAGScheduler里面调用了updateAccumulators(event)

在updateAccumulators方法中

Accumulators.add(event.accumUpdates)

具体内容如下:

def add(values: Map[Long, Any]): Unit = synchronized {
for ((id, value) <- values) {
if (originals.contains(id)) {
// Since we are now storing weak references, we must check whether the underlying data
// is valid.
originals(id).get match {
case Some(accum) => accum.asInstanceOf[Accumulable[Any, Any]] ++= value
case None =>
throw new IllegalAccessError("Attempted to access garbage collected Accumulator.")
}
} else {
logWarning(s"Ignoring accumulator update for unknown accumulator id $id")
}
}
}

5,最后我们就可以获取到累加器的值了

accum.value

五,累加器使用注意事项

累加器不会改变我们RDD的Lazy的特性,之后再Action之后完成计算和更新。

但是假如出现两个Action公用一个转化操作,如map,在map里面进行累加器累加,那么每次action都会累加,造成某些我们不需要的结果。

六,自定义累加器

自定义累加器输出

七,总结

主要牵涉点就是序列化及类加载执行,这是深入玩spark的必须.

原文地址:http://blog.51cto.com/13784648/2125352

时间: 2024-11-07 00:18:21

spark源码系列之累加器实现机制及自定义累加器的相关文章

Spark源码系列(七)Spark on yarn具体实现

本来不打算写的了,但是真的是闲来无事,整天看美剧也没啥意思.这一章打算讲一下Spark on yarn的实现,1.0.0里面已经是一个stable的版本了,可是1.0.1也出来了,离1.0.0发布才一个月的时间,更新太快了,节奏跟不上啊,这里仍旧是讲1.0.0的代码,所以各位朋友也不要再问我讲的是哪个版本,目前为止发布的文章都是基于1.0.0的代码. 在第一章<spark-submit提交作业过程>的时候,我们讲过Spark on yarn的在cluster模式下它的main class是or

Spark源码系列(四)图解作业生命周期

这一章我们探索了Spark作业的运行过程,但是没把整个过程描绘出来,好,跟着我走吧,let you know! 我们先回顾一下这个图,Driver Program是我们写的那个程序,它的核心是SparkContext,回想一下,从api的使用角度,RDD都必须通过它来获得. 下面讲一讲它所不为认知的一面,它和其它组件是如何交互的. Driver向Master注册Application过程 SparkContext实例化之后,在内部实例化两个很重要的类,DAGScheduler和TaskSched

Spark源码系列(一)spark-submit提交作业过程

前言 折腾了很久,终于开始学习Spark的源码了,第一篇我打算讲一下Spark作业的提交过程.有不明白Spark的原理的话,有另外一位大牛已经写了一个系列的Spark的源码分析了,大家可以去参考他的,他的过程图画得非常好,他写过的我可能就不写了,实在没办法比人家写得更好. 下面给出他的地址: http://www.cnblogs.com/hseagle/p/3664933.html,屌丝们,赶紧去膜拜大神吧. 这个是Spark的App运行图,它通过一个Driver来和集群通信,集群负责作业的分配

Spark源码系列(五)RDD是如何被分布式缓存?

这一章想讲一下Spark的缓存是如何实现的.这个persist方法是在RDD里面的,所以我们直接打开RDD这个类. def persist(newLevel: StorageLevel): this.type = { // StorageLevel不能随意更改 if (storageLevel != StorageLevel.NONE && newLevel != storageLevel) { throw new UnsupportedOperationException("C

Spark源码系列(三)作业运行过程

导读 看这篇文章的时候,最好是能够跟着代码一起看,我是边看代码边写的,所以这篇文章的前进过程也就是我看代码的推进过程. 作业执行 上一章讲了RDD的转换,但是没讲作业的运行,它和Driver Program的关系是啥,和RDD的关系是啥? 官方给的例子里面,一执行collect方法就能出结果,那我们就从collect开始看吧,进入RDD,找到collect方法. def collect(): Array[T] = { val results = sc.runJob(this, (iter: It

Spark源码系列(八)Spark Streaming实例分析

这一章要讲Spark Streaming,讲之前首先回顾下它的用法,具体用法请参照<Spark Streaming编程指南>. Example代码分析 val ssc = new StreamingContext(sparkConf, Seconds(1)); // 获得一个DStream负责连接 监听端口:地址 val lines = ssc.socketTextStream(serverIP, serverPort); // 对每一行数据执行Split操作 val words = line

Spark源码系列(九)Spark SQL初体验之解析过程详解

好久没更新博客了,之前学了一些R语言和机器学习的内容,做了一些笔记,之后也会放到博客上面来给大家共享.一个月前就打算更新Spark Sql的内容了,因为一些别的事情耽误了,今天就简单写点,Spark1.2马上就要出来了,不知道变动会不会很大,据说添加了很多的新功能呢,期待中... 首先声明一下这个版本的代码是1.1的,之前讲的都是1.0的. Spark支持两种模式,一种是在spark里面直接写sql,可以通过sql来查询对象,类似.net的LINQ一样,另外一种支持hive的HQL.不管是哪种方

hbase源码系列(十三)缓存机制MemStore与Block Cache

这一章讲hbase的缓存机制,这里面涉及的内容也是比较多,呵呵,我理解中的缓存是保存在内存中的特定的便于检索的数据结构就是缓存. 之前在讲put的时候,put是被添加到Store里面,这个Store是个接口,实现是在HStore里面,MemStore其实是它底下的小子. 那它和Region Server.Region是什么关系? Region Server下面有若干个Region,每个Region下面有若干的列族,每个列族对应着一个HStore. HStore里面有三个很重要的类,在这章的内容都

Android 源码系列之&lt;十四&gt;从源码的角度深入理解LeakCanary的内存泄露检测机制(下)

转载请注明出处:http://blog.csdn.net/llew2011/article/details/52958567 在上边文章Android 源码系列之<十三>从源码的角度深入理解LeakCanary的内存泄露检测机制(中)由于篇幅原因仅仅向小伙伴们讲述了在Android开发中如何使用LeakCanary来检测应用中出现的内存泄露,并简单的介绍了LeakCanary的相关配置信息.根据上篇文章的介绍我们知道LeakCanary为了不给APP进程造成影响所以新开启了一个进程,在新开启的