Spark源码学习1.5——BlockManager.scala

一、BlockResult类

该类用来表示返回的匹配的block及其相关的参数。共有三个参数:

data:Iterator [Any]。

readMethod: DataReadMethod.Value。

bytes: Long。

实例化InputMetrics类。

二、BlockManager类

关系的参数有executorId,blockManagerMaster,mapOutPutTracker等。关系的类有ShuffleBlockManager类,DiskBlockManager类,ConnectionManager类,MemoryStore类,DiskStore类等。包含以下操作:

1、检查broadcast,shuffle,rdds,shuffleSpill存储时是否被压缩。

2、定义其slaveActor。

3、建立一个BlockManagerSlaveActor(this, mapOutputTracker)。

4、一个lazy val compressionCodec,压缩包解码器。lazy表示在调用它时才实例化一个解码器,主要是针对用户自定义的jar包。

包含的方法有:

1、initialize():初始化操作,主要是向master(BlockManagerMaster)注册BlockManager,并启动BlockManagerWorker。

2、reportAllBlocks():再次将所有的blocks汇报给BlockManager。这个方法强调所有的blocks必须都能在BlockManager的管理下,因为可能会出现各种因素,如slave需要重新注册、进程冲突导致block变化等,让blocks产生变化。

3、reregister():重新注册BlockManager,这个方法主要是在心跳进程发现BlockManager没有注册时调用。这个方法的调用无需在锁状态下执行。

4、asyncReregister():异步地重新向master注册BlockManager。

5、waitForAsyncReregister():如果有其他的异步重注册进程,则等待。

6、getStatus(block Id: BlockId):根据blockId获取block的信息。

7、getMatchingBlockIds():指定过滤器对所有的blocks进行过滤。

8、reportBlockStatus():向master报告block所在存储位置的状况,这个信息不仅反映了block当前的状态,还用于更新block的信息。但是这个存储状态的信息,如磁盘、内存、cache存储等等,并不一定是block的Information中所期望的存储信息,例如MEMORY_AND_DISK等。

9、tryToReportBlockStatus():实际上是发送UpdateBlockInfo信息,返回master返回的信息。block记录正确则返回true,slave需要重新注册则返回false。信息包括storageLevel, inMemSize, inTachyonSize, onDiskSize等

10、getCurrentBlockStatus():返回指定block所在存储块的最新信息。特别的,当block从内存移到磁盘时,更改其存储级别并更新内存和磁盘大小。

11、getLocationBlockIds():获取一系列block的位置。

12、getLocalFromDisk():直接从磁盘获取block,主要是用来获取shuffle block的。

13、getLocal():从本地block manager获取block。

14、getLocakBytes():以序列化字节流的形式从本地block manager获取block。

15、doGetLocal():getLocalBytes()的实际操作函数,用来获取block并以序列化形式输出。需要处理的问题主要是同步问题,既要保证block仍然在指定的存储位置,也要保证block没有处于写的锁状态(尽管如此,Spark提示它仍然存在一定的隐患,block被移除)。

16、getRemote():从远程block manager获取block。

17、getRemoteBytes():同getLocalBytes()。

18、doGetRemote():与doGetLocal()类似。

19、get():从block manager获取block,无论本地还是远程。

20、getMultiple():根据BlockManagerIds获取一系列的block。

21、putIterator():在Iterator中加入block的信息。

22、getDiskWriter():创建一个能够直接将数据写到磁盘的writer,block通过文件名来指定写入的文件,这个方法通常用来在shuffle之后写入shuffle的输出文件。

23、putArray():在block manager中写入新的block,block的值是Array数组。

24、putBytes():在block manager中写入新的block,block的值是序列化的字符流。这个方法和putArray都是使用doPut()来实际写入信息到block manager。

25、doPut():将给定的block根据给定的存储等级(内存还是磁盘)写入相应的block存储区,必要时复制其中的数据。实际有效的存储级依赖于block最终所挂载的区域,因此调用者可以指定存储级,而不是依靠最初由用户指定的存储级来确定(用户可能存在错误或恶意行为)。doPut()需要考虑的问题包括:block信息的写入,block的值的处理,block向block manager和master的注册,block存储的位置(先内存,后实际考虑),报告block所在存储区的信息,block manager信息的更新等。

26、replicate():复制block到另外的节点上。

27、putSingle():在block manager中写入新的block,block的值是一个对象。调用的方法是putIterator。

28、dropFromMemory():将内存中的block溢出到磁盘中。往往在内存达到限制时调用。

29、removeRdd():将指定RDD的block全部移除,返回移除的block的数量。

30、removeBroadcast():将指定的broadcast的block全部移除。这个方法和removeRdd都是循环移除自身的所有block,移除的方法为removeBlock()。

31、removeBlock():移除内存和磁盘中的指定block。同时需要告知master并更新block信息reportBlockStatuses。

32、dropOldNonBroadcastBlocks()和dropOldBroadcastBlocks():移除旧的没有的/旧的broadcast block。

33、dropOldBlocks():移除没有用的block。

34、shouldCompress():判断是否经过压缩,共有四种压缩包——shuffle,broadcast,rdds,shuffleSpill。

35、wrapForCompression():有两种加载方式,根据参数决定是压缩输入流还是压缩输出流。

36、dataSerializeStram():序列化为流。

37、dataSerialize():序列化为字符缓存。

38、dateDeserialize():反序列化字符缓存为迭代的参数值,并且在末尾对该值进行处理。在对block进行处理的时候,如果block是shuffle的block,也就是会传递给reduce进行处理的block时,需要进行压缩(lazy的)。

39、stop():清除各种类的实例化对象。

三、BlockManager对象

针对blockId进行了注册、处理等操作。

时间: 2024-11-05 17:29:06

Spark源码学习1.5——BlockManager.scala的相关文章

Spark源码学习1.1——DAGScheduler.scala

本文以Spark1.1.0版本为基础. 经过前一段时间的学习,基本上能够对Spark的工作流程有一个了解,但是具体的细节还是需要阅读源码,而且后续的科研过程中也肯定要修改源码的,所以最近开始Spark的源码的学习.首先以重要文件为基础分别分析,然后再整体的分析. (一)DAGScheduler.scala文件的主要功能 DAGScheduler是划分Job为stage的调度,它是在作业所需要的数据已经被分为RDD之后执行的.DAGScheduler将Job划分为DAG图,以stage为图的结点,

Spark源码学习1.6——Executor.scala

Executor.scala 一.Executor类 首先判断本地性,获取slaves的host name(不是IP或者host: port),匹配运行环境为集群或者本地.如果不是本地执行,需要启动一个handler来监控所有的executor进程,避免阻塞.然后,初始化Spark执行环境.向SparkEnv注册executor资源,即registerSource方法.第三步,装载类,序列化类到内存中.第四,启动worker的线程池.第五,收集所有的task任务.接下就可以分配资源给task了,

Spark源码学习1.4——MapOutputTracker.scala

相关类:MapOutputTrackerMessage,GetMapOutputStatuses extends MapPutputTrackerMessage,StopMapOutputTracker extends MapOutputTrackerMessage,MapOutputTrackerMasterActor,MapOutputTrackerMaster. 首先重写MapOutputTrackerMasterActor的receiveWithLogging:判断MapOutputTr

Spark源码学习1.3——TaskSetManager.scala

TaskSetManager.scala TaskSet是指一系列被提交的task,一般是代表特定的stage中丢失的partition.TaskSetManager通过一个TaskSchedulerImpl实体来对TaskSet进行管理. 该方法定义了以下参数: EXECUTOR_TASK_BLACKLIST_TIMEOUT:executor加入黑名单的时间.如果executor意外终止导致任务执行失败,那么会暂时将这个executor加入黑名单,不再分配任务给它. SPECULATION_Q

Spark源码学习和总结1

一. Spark源码编译 解析: wget http://archive.apache.org/dist/spark/spark-1.6.0/spark-1.6.0-bin-hadoop2.6.tgz gtar -zxvf spark-1.6.0-bin-hadoop2.6.tgz cd spark-1.6.0-bin-hadoop2.6 ./sbt/sbt gen-idea 说明:漫长的等待后,上述命令生成SBT工程,我们就可以使用IDEA以SBT Project的方式打开. 二. RDD实现

Spark源码学习2

转自:http://www.cnblogs.com/hseagle/p/3673123.html 在源码阅读时,需要重点把握以下两大主线. 静态view 即 RDD, transformation and action 动态view 即 life of a job, 每一个job又分为多个stage,每一个stage中可以包含多个rdd及其transformation,这些stage又是如何映射成为task被distributed到cluster中 一.概要 本文以wordCount为例,详细说

spark源码学习-withScope

withScope是最近的发现版中新增加的一个模块,它是用来做DAG可视化的(DAG visualization on SparkUI) 以前的sparkUI中只有stage的执行情况,也就是说我们不可以看到上个RDD到下个RDD的具体信息.于是为了在 sparkUI中能展示更多的信息.所以把所有创建的RDD的方法都包裹起来,同时用RDDOperationScope 记录 RDD 的操作历史和关联,就能达成目标.下面就是一张WordCount的DAG visualization on Spark

Spark源码学习和总结2

2. SchedulerBackend SchedulerBackend是一个trait,主要的功能是向当前等待分配计算资源的Task分配计算资源Executor,并且在分配的Executor上启动Task,完成计算的调度过程.上述任务调度是通过方法def reviveOffers(): Unit实现的.需要说明的是CoarseGrainedSchedulerBackend是SchedulerBackend的一个具体实现,而YARN.Standalone和Mesos都是基于SchedulerBa

大数据Spark蘑菇云前传第15课:Scala类型参数编程实战及Spark源码鉴赏(学习笔记)

前传第15课:Scala类型参数编程实战及Spark源码鉴赏 本課課程: Spark源码中的Scala类型系統的使用 Scala类型系統编程操作实战 Spark源码中的Scala类型系統的使用 classOf[RDD[_]] 這個也是类型系統 這里的意思是說 B 這種類型必需至少是 A 這樣類型 Ordering Scala类型系統编程操作实战 作為類型系統最大的就可以對類型進行限制,在Scala 中的類型系統,他本身也作為對象.e.g. 我們可以建立 Person 這個類,現在可以建立一個什麼