Spark理论学习笔记（一）
1.调度
分为FIFO和FAIR两种模式
创建调度池：sc.setLocalProperty(“spark.scheduler.pool”, “pool6”)
终止调度池：sc.setLocalProperty(“spark.scheduler.pool6”, null)
配置调度池：
通过conf/fairscheduler.xml
sparkConf.set(“spark.scheduler.allocation.file”, “/path/to/file”)
XML文件格式参考官方。
2. 序列化、压缩
两种方式配置压缩：
1.spark-env.sh中配置：export SPARK_JAVA_OPTS=”-Dspark.broadcast.compress”
2.sparkConf.set(“spark.broadcast.compress”, true)
其它配置：
spark.broadcast.compress boolean
spark.rdd.compress boolean
spark.io.compression.codec org.apache.spark.io.LZFCompressionCodec
spark.io.compression.snappy.block.size 32768
3. Spark块管理
物理上存储RDD是以Block为单位的，一个Partition对应一个Block。Spark各节点创建各自的BlockManager
Block是由BlockManager管理的，采用了Master-Slave结构，采用Akka通信
重要概念：
Narrow dependency——窄依赖，子RDD依赖于父RDD中固定的data partition。
Wide Dependency——宽依赖，子RDD对父RDD中的所有data partition都有依赖。Spark通过宽依赖来划分Stage
延迟计算
部署（Deployment view）
当有Action作用于某RDD时，该action会作为一个job被提交。
在提交的过程中，DAGScheduler模块介入运算，计算RDD之间的依赖关系。RDD之间的依赖关系就形成了DAG。
每一个JOB被分为多个stage，划分stage的一个主要依据是当前计算因子的输入是否是确定的，如果是则将其分在同一个stage，避免多个stage之间的消息传递开销。
当stage被提交之后，由taskscheduler来根据stage来计算所需要的task，并将task提交到对应的worker。
Spark支持以下几种部署模式，Standalone、Mesos和YARN。这些部署模式将作为taskscheduler的初始化入参。
RDD接口（RDD Interface）
RDD由以下几个主要部分组成
-
partitions——partition集合，一个RDD中有多少data partition
-
dependencies——RDD依赖关系
-
compute(parition)——对于给定的数据集，需要作哪些计算
-
preferredLocations——对于data partition的位置偏好
-
partitioner——对于计算出来的数据结果如何分发
容错性（Fault-tolerant）
从最初始的RDD到衍生出来的最后一个RDD，中间要经过一系列的处理。那么如何处理中间环节出现错误的场景呢？
Spark提供的解决方案是只对失效的data partition进行事件重演，而无须对整个数据全集进行事件重演，这样可以大大加快场景恢复的开销。
RDD又是如何知道自己的data partition的number该是多少？如果是HDFS文件，那么HDFS文件的block将会成为一个重要的计算依据。
JOB的生成和运行
job生成的简单流程如下
-
首先应用程序创建SparkContext的实例，如实例为sc
-
利用SparkContext的实例来创建生成RDD
-
经过一连串的transformation操作，原始的RDD转换成为其它类型的RDD
-
当action作用于转换之后RDD时，会调用SparkContext的runJob方法
-
sc.runJob的调用是后面一连串反应的起点，关键性的跃变就发生在此处
调用路径大致如下
-
sc.runJob->dagScheduler.runJob->submitJob
-
DAGScheduler::submitJob会创建JobSummitted的event发送给内嵌类eventProcessActor
-
eventProcessActor在接收到JobSubmmitted之后调用processEvent处理函数
-
job到stage的转换，生成finalStage并提交运行，关键是调用submitStage
-
在submitStage中会计算stage之间的依赖关系，依赖关系分为宽依赖和窄依赖两种
-
如果计算中发现当前的stage没有任何依赖或者所有的依赖都已经准备完毕，则提交task
-
提交task是调用函数submitMissingTasks来完成
-
task真正运行在哪个worker上面是由TaskScheduler来管理，也就是上面的submitMissingTasks会调用TaskScheduler::submitTasks
-
TaskSchedulerImpl中会根据Spark的当前运行模式来创建相应的backend，如果是在单机运行则创建LocalBackend
-
LocalBackend收到TaskSchedulerImpl传递进来的ReceiveOffers事件
-
receiveOffers->executor.launchTask->TaskRunner.run
代码片段executor.lauchTask
def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) {
val tr = new TaskRunner(context, taskId, serializedTask)
runningTasks.put(taskId, tr)
threadPool.execute(tr)
}
最终的逻辑处理切切实实是发生在TaskRunner这么一个executor之内
Spark调度方案
Spark中的调度模式主要有两种：FIFO和FAIR。默认情况下Spark的调度模式是FIFO（先进先出），谁先提交谁先执行，后面的任务需要等待前面的任务执行。而FAIR（公平调度）模式支持在调度池中为任务进行分组，不同的调度池权重不同，任务可以按照权重来决定执行顺序。对这两种调度模式的具体实现，接下来会根据spark-1.6.0的源码来进行详细的分析。使用哪种调度器由参数spark.scheduler.mode来设置，可选的参数有FAIR和FIFO，默认是FIFO。
一、源码入口
在Scheduler模块中，当Stage划分好，然后提交Task的过程中，会进入TaskSchedulerImpl#submitTasks方法。
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties) //目前支持FIFO和FAIR两种调度策略- 1
- 1
在上面代码中有一个schedulableBuilder对象，这个对象在TaskSchedulerImpl类中的定义及实现可以参考下面这段源代码：
var schedulableBuilder: SchedulableBuilder = null
...
def initialize(backend: SchedulerBackend) {
this.backend = backend
// temporarily set rootPool name to empty
rootPool = new Pool("", schedulingMode, 0, 0)
schedulableBuilder = {
schedulingMode match {
case SchedulingMode.FIFO =>
new FIFOSchedulableBuilder(rootPool) //rootPool包含了一组TaskSetManager
case SchedulingMode.FAIR =>
new FairSchedulableBuilder(rootPool, conf) //rootPool包含了一组Pool树，这棵树的叶子节点都是TaskSetManager
}
}
schedulableBuilder.buildPools() //在FIFO中的实现是空
} 根据用户配置的SchedulingMode决定是生成FIFOSchedulableBuilder还是生成FairSchedulableBuilder类型的schedulableBuilder对象。
在生成schedulableBuilder后，调用其buildPools方法生成调度池。
调度模式由配置参数spark.scheduler.mode（默认值为FIFO）来确定。
两种模式的调度逻辑图如下：
二、FIFOSchedulableBuilder
FIFO的rootPool包含一组TaskSetManager。从上面的类继承图中看出在FIFOSchedulableBuilder中有两个方法：
1、buildPools
实现为空
override def buildPools() {
// nothing
}所以，对于FIFO模式，获取到schedulableBuilder对象后，在调用buildPools方法后，不做任何操作。
2、addTaskSetManager
该方法将TaskSetManager装载到rootPool中。直接调用的方法是Pool#addSchedulable()。
override def addTaskSetManager(manager: Schedulable, properties: Properties) { rootPool.addSchedulable(manager) }3Pool#addSchedulable()方法：
val schedulableQueue = new ConcurrentLinkedQueue[Schedulable]
...
override def addSchedulable(schedulable: Schedulable) {
require(schedulable != null)
schedulableQueue.add(schedulable)
schedulableNameToSchedulable.put(schedulable.name, schedulable)
schedulable.parent = this
}将该TaskSetManager加入到调度队列schedulableQueue中。
三、FairSchedulableBuilder
FAIR的rootPool中包含一组Pool，在Pool中包含了TaskSetManager。
1、buildPools
在该方法中，会读取配置文件，按照配置文件中的配置参数调用buildFairSchedulerPool生成配置的调度池，以及调用buildDefaultPool生成默认调度池。
默认情况下FAIR模式的配置文件是位于SPARK_HOME/conf/fairscheduler.xml文件，也可以通过参数spark.scheduler.allocation.file设置用户自定义配置文件。
spark中提供的fairscheduler.xml模板如下所示：
<allocations>
<pool name="production">
<schedulingMode>FAIR</schedulingMode>
<weight>1</weight>
<minShare>2</minShare>
</pool>
<pool name="test">
<schedulingMode>FIFO</schedulingMode>
<weight>2</weight>
<minShare>3</minShare>
</pool>
</allocations>参数含义：
（1）name: 该调度池的名称，可根据该参数使用指定pool，入sc.setLocalProperty("spark.scheduler.pool", "test")
（2）weight: 该调度池的权重，各调度池根据该参数分配系统资源。每个调度池得到的资源数为weight / sum(weight)，weight为2的分配到的资源为weight为1的两倍。
（3）minShare: 该调度池需要的最小资源数（CPU核数）。fair调度器首先会尝试为每个调度池分配最少minShare资源，然后剩余资源才会按照weight大小继续分配。
（4）schedulingMode: 该调度池内的调度模式。
2、buildFairSchedulerPool
从上面的配置文件可以看到，每一个调度池有一个name属性指定名字，然后在该pool中可以设置其schedulingMode(可为空，默认为FIFO), weight(可为空，默认值是1), 以及minShare(可为空，默认值是0)参数。然后使用这些参数生成一个Pool对象，把该pool对象放入rootPool中。入下所示：
val pool = new Pool(poolName, schedulingMode, minShare, weight)
rootPool.addSchedulable(pool)3、buildDefaultPool
如果如果配置文件中没有设置一个name为default的pool，系统才会自动生成一个使用默认参数生成的pool对象。各项参数的默认值在buildFairSchedulerPool中有提到。
4、addTaskSetManager
这一段逻辑中是把配置文件中的pool，或者default pool放入rootPool中，然后把TaskSetManager存入rootPool对应的子pool。
override def addTaskSetManager(manager: Schedulable, properties: Properties) {
var poolName = DEFAULT_POOL_NAME
var parentPool = rootPool.getSchedulableByName(poolName)
if (properties != null) {
poolName = properties.getProperty(FAIR_SCHEDULER_PROPERTIES, DEFAULT_POOL_NAME)
parentPool = rootPool.getSchedulableByName(poolName)
if (parentPool == null) {
// we will create a new pool that user has configured in app
// instead of being defined in xml file
parentPool = new Pool(poolName, DEFAULT_SCHEDULING_MODE,
DEFAULT_MINIMUM_SHARE, DEFAULT_WEIGHT)
rootPool.addSchedulable(parentPool)
logInfo("Created pool %s, schedulingMode: %s, minShare: %d, weight: %d".format(
poolName, DEFAULT_SCHEDULING_MODE, DEFAULT_MINIMUM_SHARE, DEFAULT_WEIGHT))
}
}
parentPool.addSchedulable(manager)
logInfo("Added task set " + manager.name + " tasks to pool " + poolName)
}5、FAIR调度池使用方法
在Spark-1.6.1官方文档中写道：
如果不加设置，jobs会提交到default调度池中。由于调度池的使用是Thread级别的，只能通过具体的SparkContext来设置local属性（即无法在配置文件中通过参数
spark.scheduler.pool来设置，因为配置文件中的参数会被加载到SparkConf对象中）。所以需要使用指定调度池的话，需要在具体代码中通过SparkContext对象sc来按照如下方法进行设置：sc.setLocalProperty("spark.scheduler.pool", "test")
设置该参数后，在该thread中提交的所有job都会提交到test Pool中。
如果接下来不再需要使用到该test调度池，sc.setLocalProperty("spark.scheduler.pool", null)
四、FIFO和FAIR的调度顺序
这里必须提到的一个类是上面提到的Pool，在这个类中实现了不同调度模式的调度算法。
var taskSetSchedulingAlgorithm: SchedulingAlgorithm = {
schedulingMode match {
case SchedulingMode.FAIR =>
new FairSchedulingAlgorithm()
case SchedulingMode.FIFO =>
new FIFOSchedulingAlgorithm()
}
}FIFO模式的算法类是FIFOSchedulingAlgorithm，FAIR模式的算法实现类是FairSchedulingAlgorithm。
接下来的两节中comparator方法传入参数Schedulable类型是一个trait，具体实现主要有两个：1，Pool；2，TaskSetManager。与最前面那个调度模式的逻辑图相对应。
1、FIFO模式的调度算法FIFOSchedulingAlgorithm
在这个类里面，主要逻辑是一个comparator方法。
override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
val priority1 = s1.priority //实际上是Job ID
val priority2 = s2.priority
var res = math.signum(priority1 - priority2)
if (res == 0) { //如果Job ID相同，就比较Stage ID
val stageId1 = s1.stageId
val stageId2 = s2.stageId
res = math.signum(stageId1 - stageId2)
}
if (res < 0) {
true
} else {
false
}
}如果有两个调度任务s1和s2，首先获得两个任务的priority，在FIFO中该优先级实际上是Job ID。首先比较两个任务的Job ID，如果priority1比priority2小，那么返回true，表示s1的优先级比s2的高。我们知道Job ID是顺序生成的，先生成的Job ID比较小，所以先提交的job肯定比后提交的job先执行。但是如果是同一个job的不同任务，接下来就比较各自的Stage ID，类似于比较Job ID，Stage ID小的优先级高。
2、FAIR模式的调度算法FairSchedulingAlgorithm
这个类中的comparator方法源代码如下：
override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
val minShare1 = s1.minShare //在这里share理解成份额，即每个调度池要求的最少cpu核数
val minShare2 = s2.minShare
val runningTasks1 = s1.runningTasks // 该Pool或者TaskSetManager中正在运行的任务数
val runningTasks2 = s2.runningTasks
val s1Needy = runningTasks1 < minShare1 // 如果正在运行任务数比该调度池最小cpu核数要小
val s2Needy = runningTasks2 < minShare2
val minShareRatio1 = runningTasks1.toDouble / math.max(minShare1, 1.0).toDouble
val minShareRatio2 = runningTasks2.toDouble / math.max(minShare2, 1.0).toDouble
val taskToWeightRatio1 = runningTasks1.toDouble / s1.weight.toDouble
val taskToWeightRatio2 = runningTasks2.toDouble / s2.weight.toDouble
var compare: Int = 0
if (s1Needy && !s2Needy) {
return true
} else if (!s1Needy && s2Needy) {
return false
} else if (s1Needy && s2Needy) {
compare = minShareRatio1.compareTo(minShareRatio2)
} else {
compare = taskToWeightRatio1.compareTo(taskToWeightRatio2)
}
if (compare < 0) {
true
} else if (compare > 0) {
false
} else {
s1.name < s2.name
}
} minShare对应fairscheduler.xml配置文件中的minShare属性。
（1）如果s1所在Pool或者TaskSetManager中运行状态的task数量比minShare小，s2所在Pool或者TaskSetManager中运行状态的task数量比minShare大，那么s1会优先调度。反之，s2优先调度。
（2）如果s1和s2所在Pool或者TaskSetManager中运行状态的task数量都比各自minShare小，那么minShareRatio小的优先被调度。
minShareRatio是运行状态task数与minShare的比值，即相对来说minShare使用较少的先被调度。
（3）如果minShareRatio相同，那么最后比较各自Pool的名字。
Spark内存分配
SPARK的内存管理器
StaticMemoryManager,UnifiedMemoryManager
1.6以后默认是UnifiedMemoryManager.
这个内存管理器在sparkContext中通过SparnEnv.create函数来创建SparkEnv的实例时,会生成.
通过Spark.memory.useLegacyMode配置,可以控制选择的内存管理器实例.
如果设置为true时,选择的实例为StaticMemoryManager实例,否则选择UnifiedMemoryManager实例.默认情况下这个值为false.
val useLegacyMemoryManager = conf.getBoolean(“spark.memory.useLegacyMode”, false)
val memoryManager: MemoryManager =
if (useLegacyMemoryManager) {
new StaticMemoryManager(conf, numUsableCores)
} else {
UnifiedMemoryManager(conf, numUsableCores)
}
UnifiedMemoryManager
这个实例生成时,最大内存的得到方法:
1,根据当前JVM的启动内存,减去300MB,
这个300MB可以通过spark.testing.reservedMemory配置得到.
2,最大内存值,通过1计算出来的内存值,与spark.memory.fraction配置的系数进行相乘.默认是0.75.
示例:如果JVM配置的内存为1GB,那么可使用的最大内存为(1GB-300MB)*0.75
需要的配置项:
配置项spark.memory.fraction,默认值0.75,这个配置用于配置当前的内存管理器的最大内存使用比例.
配置项spark.memory.storageFraction,默认值0.5,这个配置用于配置rdd的storage与cache的默认分配的内存池大小.
配置项spark.memory.offHeap.size,默认值0,这个配置用于配置非堆内存的大小,默认不启用.这个不做分析.
在实例生成后,默认会根据storage的内存权重,总内存减去storage的内存权重,生成两个内存池storageMemoryPool与onHeapExecutionMemoryPool.
onHeapExecutionMemoryPool用于在执行executor的shuffle操作时,使用的内存,
storageMemoryPool用于在执行rdd的cache操作时,使用的内存.
在Executor执行时的内存分配
这个操作通常是在task执行shuffle操作时,计算spill时,在内存中的CACHE时使用的内存.通过调用实例中的acquireExecutionMemory函数来申请内存.
override private[memory] def acquireExecutionMemory(
numBytes: Long,
taskAttemptId: Long,
memoryMode: MemoryMode): Long = synchronized {
这个函数传入的memoryMode可选择是使用堆内存还是直接使用本地内存,默认是使用堆内存.
assert(onHeapExecutionMemoryPool.poolSize +
storageMemoryPool.poolSize == maxMemory)
assert(numBytes >= 0)
memoryMode match {
case MemoryMode.ON_HEAP =>
这里定义的这个函数,用于判断numBytes(需要申请的内存大小)减去当前内存池中可用的内存大小是否够用,如果不够用,这个函数的传入值是一个正数
def maybeGrowExecutionPool(extraMemoryNeeded: Long): Unit = {
if (extraMemoryNeeded > 0) {
这里根据当前的rdd的cache中的内存池的大小,减去配置的storage的存储大小,与当前storage的内存池中的可用大小,取最大值出来,这个值表示是一个可用于回收的内存资源.
val memoryReclaimableFromStorage =
math.max(storageMemoryPool.memoryFree,
storageMemoryPool.poolSize – storageRegionSize)
if (memoryReclaimableFromStorage > 0) {
首先根据计算出来的storage中可以进行回收的资源,通过StorageMemoryPool进行资源的释放.得到一个完成释放的资源大小.这里根据executor中task需要的内存与storage可回收的资源取最小值进行资源的回收.把得到的可用资源添加到executor的内存池中.
// Only reclaim as much space as is necessary and available:
val spaceReclaimed = storageMemoryPool.shrinkPoolToFreeSpace(
math.min(extraMemoryNeeded, memoryReclaimableFromStorage))
onHeapExecutionMemoryPool.incrementPoolSize(spaceReclaimed)
}
}
}
这个函数用于计算executor的内存池可以使用的最大内存大小.最小可以使用总内存减去storage的配置权重,也就是默认情况下,shuffle的executor的内存最小可以使用0.5的权重的内存.
def computeMaxExecutionPoolSize(): Long = {
maxMemory – math.min(storageMemoryUsed, storageRegionSize)
}
执行内存的分配操作.
onHeapExecutionMemoryPool.acquireMemory(
numBytes, taskAttemptId, maybeGrowExecutionPool,
computeMaxExecutionPoolSize)
case MemoryMode.OFF_HEAP =>
offHeapExecutionMemoryPool.acquireMemory(numBytes, taskAttemptId)
}
}
给executor中的task分配需要的内存:
private[memory] def acquireMemory(
numBytes: Long,
taskAttemptId: Long,
maybeGrowPool: Long => Unit = (additionalSpaceNeeded: Long) => Unit,
computeMaxPoolSize: () => Long = () => poolSize): Long = lock.synchronized {
assert(numBytes > 0, s”invalid number of bytes requested: $numBytes”)
// TODO: clean up this clunky method signature
if (!memoryForTask.contains(taskAttemptId)) {
这里首先检查这个task是否在memoryForTask中存在,如果不存在时,表示这个task是第一次申请内存,在这个集合中设置此task的当前使用内存为0,并唤醒所有的当前的executor的等待的task.
memoryForTask(taskAttemptId) = 0L
// This will later cause waiting tasks to wake up and check numTasks again
lock.notifyAll()
}
// TODO: simplify this to limit each task to its own slot
while (true) {
执行内存的分配操作,这个操作会一直进行迭代,直到满足一定的条件.
首先得到当前的executor中有申请内存的task的个数,并得到当前的task的使用内存量.
val numActiveTasks = memoryForTask.keys.size
val curMem = memoryForTask(taskAttemptId)
计算出需要申请的内存与当前内存池中的内存,是否需要对storage中的内存进行回收.如果需要申请的内存大于了当前内存池的内存,这个参数传入为一个大于0的数,这个时候会对storage的内存进行回收.
maybeGrowPool(numBytes – memoryFree)
这里计算出executor的内存值可以使用的最大内存,默认情况下,最小可使用内存为总内存减去storage的配置内存.也就是默认可使用50%的内存.
val maxPoolSize = computeMaxPoolSize()
这里计算出每个task平均可使用的最大内存大小,与最小内存大小.
如:有5个task,可使用100MB的内存,那么最大可使用的内存为20MB,最小可使用的内存为10MB.
val maxMemoryPerTask = maxPoolSize / numActiveTasks
val minMemoryPerTask = poolSize / (2 * numActiveTasks)
这里计算出当前可申请的内存.能够申请的内存总量不能超过平均每个task使用内存的平均大小.
// How much we can grant this task; keep its share within 0 <= X <= 1 / numActiveTasks
val maxToGrant = math.min(numBytes, math.max(0, maxMemoryPerTask – curMem))
val toGrant = math.min(maxToGrant, memoryFree)
这里控制迭代是否可以跳出的条件.如果可申请的内存小于需要申请的内存,同时当前task使用的内存加上可申请的内存小于每个task平均使用的内存时,这个申请操作会wait住.等待其它的task资源回收时进行唤醒.否则跳出迭代,返回可申请的内存.
if (toGrant < numBytes && curMem + toGrant < minMemoryPerTask) {
logInfo(s”TID $taskAttemptId waiting for at least 1/2N of
$poolName pool to be free”)
lock.wait()
} else {
memoryForTask(taskAttemptId) += toGrant
return toGrant
}
}
0L // Never reached
}
在BLOCK的CACHE时的内存分配
在执行rdd的iterator操作时,如果对rdd执行过cache或者persist的操作时,也就是storage的级别不是none时,会对数据进行cache操作.在cache后的block中有一个超时时间,这个超时时间在blockManager中通过一个定时器,会定时去删除cache的block与的Broadcast数据.
如果是BLOCK的CACHE的超时,可通过spark.cleaner.ttl.BLOCK_MANAGER配置.
在对RDD执行cache操作时,最终会调用内存管理器中的acquireStorageMemory函数来进行操作.
override def acquireStorageMemory(
blockId: BlockId,
numBytes: Long,
evictedBlocks: mutable.Buffer[(BlockId, BlockStatus)])
: Boolean = synchronized {
这个传入的参数中,evictedBlocks是一个用于返回的传入参数,这个集合中表示进行这次申请后,被淘汰掉的block的信息.
assert(onHeapExecutionMemoryPool.poolSize
+ storageMemoryPool.poolSize == maxMemory)
assert(numBytes >= 0)
如果当前的BLOCK的的CACHE的大小已经超过了当前可用的内存总量(总内存减去executor的使用内存)时,直接返回false,表示不做内存分配.申请的内存太多,不处理了.
if (numBytes > maxStorageMemory) {
// Fail fast if the block simply won\’t fit
logInfo(s”Will not store $blockId as the required space
($numBytes bytes) exceeds our ” +
s”memory limit ($maxStorageMemory bytes)”)
return false
}
如果当前申请的block需要的内存大小超过了当前storage的内存池可用的内存大小时,在executor的内存池中回收部分资源,原则是如果申请的内存小于executor内存池可用的内存,回收申请的大小,否则回收executor所有的可用内存.并执行内存的分配操作.
if (numBytes > storageMemoryPool.memoryFree) {
// There is not enough free memory in the storage pool,
// so try to borrow free memory from
// the execution pool.
val memoryBorrowedFromExecution =
Math.min(onHeapExecutionMemoryPool.memoryFree, numBytes)
onHeapExecutionMemoryPool.decrementPoolSize(memoryBorrowedFromExecution)
storageMemoryPool.incrementPoolSize(memoryBorrowedFromExecution)
}
storageMemoryPool.acquireMemory(blockId, numBytes, evictedBlocks)
}
在StorageMemoryPool中执行block的内存分配:
def acquireMemory(
blockId: BlockId,
numBytesToAcquire: Long,
numBytesToFree: Long,
evictedBlocks: mutable.Buffer[(BlockId, BlockStatus)])
: Boolean = lock.synchronized {
第二个参数是需要申请的内存,第三个参数如果是0表示可用内存大于申请内存,大于0表示可用内存不够用.
assert(numBytesToAcquire >= 0)
assert(numBytesToFree >= 0)
assert(memoryUsed <= poolSize)
这里,如果可用的内存不够用时,通过MemoryStore中的evictBlocksToFreeSpace函数来对当前的cache进行淘汰.
if (numBytesToFree > 0) {
memoryStore.evictBlocksToFreeSpace(Some(blockId), numBytesToFree,
evictedBlocks)
// Register evicted blocks, if any, with the active task metrics
Option(TaskContext.get()).foreach { tc =>
val metrics = tc.taskMetrics()
val lastUpdatedBlocks = metrics.updatedBlocks.getOrElse(Seq[(BlockId,
BlockStatus)]())
metrics.updatedBlocks = Some(lastUpdatedBlocks ++ evictedBlocks.toSeq)
}
}
如果完成资源的回收后,当前可用的内存大于要申请的内存,表示申请成功,返回的值为true,否则为false.
// NOTE: If the memory store evicts blocks,
// then those evictions will synchronously call
// back into this StorageMemoryPool in order to free memory. Therefore,
// these variables
// should have been updated.
val enoughMemory = numBytesToAcquire <= memoryFree
if (enoughMemory) {
_memoryUsed += numBytesToAcquire
}
enoughMemory
}
Block的cache的淘汰
在executor的内存池不够使用时,或者总的内存不够使用时,会执行storage的内存池的资源回收操作.由shrinkPoolToFreeSpace函数,这个函数通过调用MemoryStorage中的evictBlocksToFreeSpace函数来进行block的淘汰(如果是对block的cache时,申请内存不够时会直接调用这个函数来淘汰老的block)
shrinkPoolToFreeSpace函数用于在executor的内存不够时,需要storage的内存池释放资源给executor使用时调用.
这个过程中,可给executor提供的内存分为五种可能:
1,storage默认的内存空间还没有使用完成,同时executor需要的空间小于等于storage的内存池的可用空间,直接在storage的内存池中释放需要的大小.
2,storage默认的内存空间还没有使用完成,同时executor需要的空间大于storage的内存池的可用空间,这个时候storage的可用空间全部进行释放,但这个时候不会做block的淘汰操作.
3,storage的默认的内存空间使用完成,这个时候storage的内存池比默认的storage的配置权重要多,同时executor需要申请的内存小于多出的部分,对storage内存池中的block进行淘汰直到够executor的申请内存结束,这个时候storage的使用内存还是大于storage的默认配置权重大小.
4,storage的默认的内存空间使用完成,这个时候storage的内存池比默认的storage的配置权重要多,同时executor需要申请的内存大于或等于多出的部分,对storage内存池中的block进行淘汰直到但最多只淘汰到storage的配置权重大小就结束淘汰.
5,storage刚好使用到了配置的权重,无法进行分配.
def shrinkPoolToFreeSpace(spaceToFree: Long): Long = lock.synchronized {
首先根据需要释放的内存,1=executor申请的内存,2-1=storage内存池可用的内存,2-2=storage中占用的内存大于默认给storage分配的权重.
这里根据这个要释放的资源与内存池可用的资源取最小值进行释放,如果申请的小于可用的,不会对block进行淘汰操作,否则对block进行淘汰操作,直接淘汰到可用的内存空间结束.
// First, shrink the pool by reclaiming free memory:
val spaceFreedByReleasingUnusedMemory = math.min(spaceToFree, memoryFree)
decrementPoolSize(spaceFreedByReleasingUnusedMemory)
val remainingSpaceToFree = spaceToFree – spaceFreedByReleasingUnusedMemory
if (remainingSpaceToFree > 0) {
// If reclaiming free memory did not adequately shrink the pool,
// begin evicting blocks:
val evictedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]
memoryStore.evictBlocksToFreeSpace(None, remainingSpaceToFree, evictedBlocks)
val spaceFreedByEviction = evictedBlocks.map(_._2.memSize).sum
// When a block is released, BlockManager.dropFromMemory()
// calls releaseMemory(), so we do
// not need to decrement _memoryUsed here. However, we do need to decrement the
// pool size.
decrementPoolSize(spaceFreedByEviction)
spaceFreedByReleasingUnusedMemory + spaceFreedByEviction
} else {
spaceFreedByReleasingUnusedMemory
}
}
evictBlocksToFreeSpace函数这个函数用于对storage的内存空间中释放掉部分block的存储空间的函数,由MemoryStorage进行实现.
这个函数的三个传入参数中:
第一个在block的cache时,会传入blockid,如果是executor要求释放时,传入为None,这个参数用于控制释放的资源,如果传入了blockid,那么这个block对应的rdd的所有的所有的CACHE都会被保留,只释放其它的RDD对应的BLOCK的CACHE,如果传入为None时,不区分BLOCK,从头开始迭代,直接释放到需要的内存大小结束.
第二个是需要释放的内存大小.
第三个参数是释放后的block的集合,这个集合内容就是从内存中淘汰出去的block.
private[spark] def evictBlocksToFreeSpace(
blockId: Option[BlockId],
space: Long,
droppedBlocks: mutable.Buffer[(BlockId, BlockStatus)]): Boolean = {
assert(space > 0)
memoryManager.synchronized {
var freedMemory = 0L
这里得到传入的blockid对应的rdd.如果传入的blockId是None时,这个rdd也就不存在.
val rddToAdd = blockId.flatMap(getRddId)
val selectedBlocks = new ArrayBuffer[BlockId]
// This is synchronized to ensure that the set of entries is not changed
// (because of getValue or getBytes) while traversing the iterator, as that
// can lead to exceptions.
entries.synchronized {
val iterator = entries.entrySet().iterator()
这里从所有的cache的block中进行迭代,如果迭代的block的rdd不是现在需要cache的block对应的rdd时(传入的blockId对应的RDD),就选择这个block.并释放内存大小.
while (freedMemory < space && iterator.hasNext) {
val pair = iterator.next()
val blockId = pair.getKey
if (rddToAdd.isEmpty || rddToAdd != getRddId(blockId)) {
selectedBlocks += blockId
freedMemory += pair.getValue.size
}
}
}
if (freedMemory >= space) {
logInfo(s”${selectedBlocks.size} blocks selected for dropping”)
for (blockId <- selectedBlocks) {
这里对选择的block,通过blockManager释放掉block的cache占用的内存.如果这个block的cache的级别中包含有disk的级别时,释放掉内存的同时会把这个cache的数据写入到磁盘中.
把执行释放后的block的集合添加到传入参数的droppedBlocks的集合参数中,用于数据的返回.
val entry = entries.synchronized { entries.get(blockId) }
// This should never be null as only one task should be dropping
// blocks and removing entries. However the check is still here for
// future safety.
if (entry != null) {
val data = if (entry.deserialized) {
Left(entry.value.asInstanceOf[Array[Any]])
} else {
Right(entry.value.asInstanceOf[ByteBuffer].duplicate())
}
val droppedBlockStatus = blockManager.dropFromMemory(blockId, data)
droppedBlockStatus.foreach { status => droppedBlocks +=
((blockId, status)) }
}
}
true
} else {
blockId.foreach { id =>
logInfo(s”Will not store $id as it would require dropping another block ” +
“from the same RDD”)
}
false
}
}
}
StaticMemoryManager
这个实例在1.6的环境下,需要把配置项spark.memory.useLegacyMode设置为true时,才会被启用.下面首先先看看这个实例生成时的处理:
需要的配置项:
1,配置项spark.shuffle.memoryFraction,用于设置executor的shuffle操作可使用的内存,默认占总内存的0.2.
2,配置项spark.shuffle.safetyFraction,用于设置executor的shuffle的安全操作内存,默认占1配置内存的0.8.
3,配置项spark.storage.memoryFraction,用于设置block cache的使用内存,默认占总内存的0.6;
4,配置项spark.storage.safetyFraction,用于设置block cache的安全使用内存,默认占3配置内存的0.9;
5,配置项spark.storage.unrollFraction,默认值是storage内存总大于的0.2;这个有于在storage中cache的block的数据的反序列化时数据的展开使用空间.
def this(conf: SparkConf, numCores: Int) {
this(
conf,
StaticMemoryManager.getMaxExecutionMemory(conf),
StaticMemoryManager.getMaxStorageMemory(conf),
numCores)
}
在Executor执行时的内存分配
在StaticMemoryManager中,对executor中的shuffle的内存执行分配这块其实并没有统一内存管理中那么麻烦,只是在分配的固定大小的存储空间中进行分配,如果无法再进行分配时,这个分配函数返回的分配量就是0.
private[memory] override def acquireExecutionMemory(
numBytes: Long,
taskAttemptId: Long,
memoryMode: MemoryMode): Long = synchronized {
OFF_HEAP的模式这里就不分析了,我在代码里没发现有地方去调用,好像申请内存时,是直接写死的ON_HEAP的模式.在这个地方,不会考虑executor的内存池中的内存是否够用,直接通过ExecutionMemoryPool内存池实例中的分配内存函数进行内存的分配 .
memoryMode match {
case MemoryMode.ON_HEAP => onHeapExecutionMemoryPool.acquireMemory(numBytes,
taskAttemptId)
case MemoryMode.OFF_HEAP => offHeapExecutionMemoryPool.acquireMemory(numBytes,
taskAttemptId)
}
}
内存分配部分的代码实现:这个部分与统一内存管理部分是一样的,
private[memory] def acquireMemory(
numBytes: Long,
taskAttemptId: Long,
maybeGrowPool: Long => Unit = (additionalSpaceNeeded: Long) => Unit,
computeMaxPoolSize: () => Long = () => poolSize): Long = lock.synchronized {
assert(numBytes > 0, s”invalid number of bytes requested: $numBytes”)
// TODO: clean up this clunky method signature
首先如果说task是第一次申请内存,添加这个task到内存池的集合属性中,并把这个task的使用内存设置为0.
if (!memoryForTask.contains(taskAttemptId)) {
memoryForTask(taskAttemptId) = 0L
// This will later cause waiting tasks to wake up and check numTasks again
lock.notifyAll()
}
下面开始迭代进行内存的分配.加上while的目的是为了保持如果task的分配内存达不到指定的大小时,就一直等待分配,直到达到指定的大小.
// TODO: simplify this to limit each task to its own slot
while (true) {
val numActiveTasks = memoryForTask.keys.size
val curMem = memoryForTask(taskAttemptId)
在这里,这个函数是一个空的实现,什么都不会做.
maybeGrowPool(numBytes – memoryFree)
这个函数得到的值就是当前的executor的内存池的poolsize的大小.
val maxPoolSize = computeMaxPoolSize()
根据当前的活动的task的个数计算出每个task可使用的最大内存,每个task使用的最小内存为最大内存除以2(如果申请的内存本身小于这个最小内存除外).
val maxMemoryPerTask = maxPoolSize / numActiveTasks
val minMemoryPerTask = poolSize / (2 * numActiveTasks)
计算出这次需要分配的内存,如果申请的内存小于可用的内存时,取申请内存,否则取这个task可申请的最大内存
// How much we can grant this task; keep its share within 0 <= X <= 1 / numActiveTasks
val maxToGrant = math.min(numBytes, math.max(0, maxMemoryPerTask – curMem))
这里计算出来的值根据当前原则上可以申请的内存与当前内存池中的可用内存取最小值.
val toGrant = math.min(maxToGrant, memoryFree)
这里有一个线程wait的条件,如果这一次申请的内存小于需要申请的内存,同时当前的task的使用内存小于最小的使用内存时,线程wait,等待其它的task释放内存或者有新的task加入来唤醒此wait.
// We want to let each task get at least 1 / (2 * numActiveTasks) before blocking;
// if we can\’t give it this much now, wait for other tasks to free up memory
// (this happens if older tasks allocated lots of memory before N grew)
if (toGrant < numBytes && curMem + toGrant < minMemoryPerTask) {
logInfo(s”TID $taskAttemptId waiting for at least 1/2N of
$poolName pool to be free”)
lock.wait()
} else {
memoryForTask(taskAttemptId) += toGrant
return toGrant
}
}
0L // Never reached
}
Storage的展开内存分配
这里说明下,在UnifiedMemoryManager中展开内存的分配与stroage中block cache的内存分配共用相同的内存空间,因此申请方法与storage的block cache的内存分配相同,而在static的分配中,不同的区块,所使用的内存空间都是固定的,因此这里需要独立说明一下.
在对MemoryStorage中执行block的cache操作时,会执行pubInterator等操作,会先根据block中的数据申请对应数据大小的展开内存空间,把数据进行提取,然后才会执行storage的cache操作的内存分配.
执行流程,在MemoryStorage中:
1,putIterator函数执行,对block进行cache
2,unrollSafely函数执行,申请展开内存,根据block的内容大小.
3,释放申请的展开内存,并申请block cache内存,执行putArray->tryToPut函数.
看看unrollSafely函数如何处理展开内存的申请:
这里先配置项spark.storage.unrollMemoryThreshold,默认值是1MB,先申请固定大小的展开内存,这个函数返回的值是一个true/false的值,true表示申请成功.这个函数调用内存管理器中的acquireUnrollMemory函数.这里申请到的内存大小会先记录到unrollMemoryMap集合中根据对应的taskid.
keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold,
droppedBlocks)
接下来,迭代block中的数据,把数据添加到vector的展开临时变量中.
while (values.hasNext && keepUnrolling) {
vector += values.next()
if (elementsUnrolled % memoryCheckPeriod == 0) {
// If our vector\’s size has exceeded the threshold, request more memory
val currentSize = vector.estimateSize()
if (currentSize >= memoryThreshold) {
这里每次申请当前的使用内存的一半做为展开内存,这个展开内存伴随着block的数据越多,申请的量也会越大.
val amountToRequest = (currentSize * memoryGrowthFactor –
memoryThreshold).toLong
keepUnrolling = reserveUnrollMemoryForThisTask(
blockId, amountToRequest, droppedBlocks)
// New threshold is currentSize * memoryGrowthFactor
memoryThreshold += amountToRequest
}
}
elementsUnrolled += 1
}
这里的判断比较关键,如果keepUnrolling的值为true,表示内存能够安全展开这个block的数据,否则表示不能展开这个block的内容.
if (keepUnrolling) {
// We successfully unrolled the entirety of this block
Left(vector.toArray)
} else {
// We ran out of space while unrolling the values for this block
logUnrollFailureMessage(blockId, vector.estimateSize())
Right(vector.iterator ++ values)
}
if (keepUnrolling) {
val taskAttemptId = currentTaskAttemptId()
memoryManager.synchronized {
这里如果内存能够安全展开当前的block,把这个block的展开内存存储到pendingUnrollMemoryMap的集合中对应此task的位置.
// Since we continue to hold onto the array until we actually cache it, we cannot
// release the unroll memory yet. Instead, we transfer it to pending unroll memory
// so `tryToPut` can further transfer it to normal storage memory later.
// TODO: we can probably express this without pending unroll memory (SPARK-10907)
val amountToTransferToPending = currentUnrollMemoryForThisTask –
previousMemoryReserved
unrollMemoryMap(taskAttemptId) -= amountToTransferToPending
pendingUnrollMemoryMap(taskAttemptId) =
pendingUnrollMemoryMap.getOrElse(taskAttemptId, 0L) +
amountToTransferToPending
}
}
提示:关于展开内存的释放部分,如果block的内容能够被安全展开存储到内存中时,这个时候,在做block的storage的操作时,会释放掉展开内存的空间(在pendingUnrollMemoryMap集合中),如果内存不能够安全展开block的内容时,这个时候无法进行block的cache操作(可能会写磁盘),这时申请的内容大小存储在unrollMemoryMap集合中,这时由于不会执行block的memory的cache操作,因此这个集合中占用的内存大小暂时不会被回收,只有等到这个task结束时,占用的unrollMemoryMap集合中的内存才会被回收.
…
接下来看看在StaticMemoryManager中如何处理展开内存的分配:
override def acquireUnrollMemory(
blockId: BlockId,
numBytes: Long,
evictedBlocks: mutable.Buffer[(BlockId, BlockStatus)])
: Boolean = synchronized {
val currentUnrollMemory = storageMemoryPool.memoryStore.currentUnrollMemory
val freeMemory = storageMemoryPool.memoryFree
这里根据可用的最大展开内存与当前正在使用中的展开内存,计算出可以申请的最大展开内存,如果这里得到的值是一个0时,表示不需要释放block cache的内存,如果是一个大于0的值,就表示需要释放BLOCK CACHE的内存.
// When unrolling, we will use all of the existing free memory, and, if necessary,
// some extra space freed from evicting cached blocks. We must place a cap on the
// amount of memory to be evicted by unrolling, however, otherwise unrolling one
// big block can blow away the entire cache.
val maxNumBytesToFree = math.max(0, maxUnrollMemory – currentUnrollMemory –
freeMemory)
这里计算出需要释放的内存,取申请的资源与可以使用的unroll内存资源的最小值,如果这个一个大于0的值,表示需要从storage的内存池中释放这么多的内存出来.
// Keep it within the range 0 <= X <= maxNumBytesToFree
val numBytesToFree = math.max(0, math.min(maxNumBytesToFree,
numBytes – freeMemory))
storageMemoryPool.acquireMemory(blockId, numBytes, numBytesToFree,
evictedBlocks)
}
Storage中block cache的内存分配
在block使用了memory的storage时,同时block的内容能够被展开内存存储起来时,会通过MemoryStorage中对应的函数来向StaticMemoryManager中的acquireStorageMemory函数申请内存资源.
override def acquireStorageMemory(
blockId: BlockId,
numBytes: Long,
evictedBlocks: mutable.Buffer[(BlockId, BlockStatus)])
: Boolean = synchronized {
if (numBytes > maxStorageMemory) {
如果一个block的内容太大,已经超过了配置的storage的存储空间大小,这个block不做cache.
// Fail fast if the block simply won\’t fit
logInfo(s”Will not store $blockId as the required space
($numBytes bytes) exceeds our ” +
s”memory limit ($maxStorageMemory bytes)”)
false
} else {
否则通过storage的内存池执行block的cache的内存申请,这个过程中如果内存不够用时,会释放老的block的cache对应的内存空间,也就是会淘汰掉老的block cache,
storageMemoryPool.acquireMemory(blockId, numBytes, evictedBlocks)
}
}
在storage的内存池中处理block cache的内存申请:
def acquireMemory(
blockId: BlockId,
numBytesToAcquire: Long,
numBytesToFree: Long,
evictedBlocks: mutable.Buffer[(BlockId, BlockStatus)])
: Boolean = lock.synchronized {
这个函数的传入参数中,numBytesToAcquire表示需要申请的内存大小,numBytesToFree如果是一个大于0的值,表示现在内存池中的内存空间不够,需要淘汰现有的block的cache.
assert(numBytesToAcquire >= 0)
assert(numBytesToFree >= 0)
assert(memoryUsed <= poolSize)
这里先判断,如果申请的内存大于还在可用的内存,需要先淘汰掉部分block cache来释放空间.
if (numBytesToFree > 0) {
memoryStore.evictBlocksToFreeSpace(Some(blockId), numBytesToFree,
evictedBlocks)
// Register evicted blocks, if any, with the active task metrics
Option(TaskContext.get()).foreach { tc =>
val metrics = tc.taskMetrics()
val lastUpdatedBlocks = metrics.updatedBlocks.getOrElse(Seq[(BlockId,
BlockStatus)]())
metrics.updatedBlocks = Some(lastUpdatedBlocks ++ evictedBlocks.toSeq)
}
}
分配内存是否成功,也就是要申请的内存小于或等于可用的内存空间,最后把分配的内存添加到使用的内存空间中.表示这部分内存已经被look住.
val enoughMemory = numBytesToAcquire <= memoryFree
if (enoughMemory) {
_memoryUsed += numBytesToAcquire
}
enoughMemory
}
Storage处理block cache的淘汰
在storage中内存不够使用时,通过memoryStorage去执行block的淘汰,并把淘汰后的block返回通知上层的调用端.
if (numBytesToFree > 0) {
memoryStore.evictBlocksToFreeSpace(Some(blockId), numBytesToFree, evictedBlocks)
// Register evicted blocks, if any, with the active task metrics
Option(TaskContext.get()).foreach { tc =>
val metrics = tc.taskMetrics()
val lastUpdatedBlocks = metrics.updatedBlocks.getOrElse(Seq[(BlockId,
BlockStatus)]())
metrics.updatedBlocks = Some(lastUpdatedBlocks ++ evictedBlocks.toSeq)
}
}
MemoryStore中处理对cache的淘汰:
对block的cache进行淘汰的处理函数,传入参数中,第二个参数是需要释放的空间,第三个参数是被淘汰后的block的集合用于返回.
private[spark] def evictBlocksToFreeSpace(
blockId: Option[BlockId],
space: Long,
droppedBlocks: mutable.Buffer[(BlockId, BlockStatus)]): Boolean = {
assert(space > 0)
memoryManager.synchronized {
var freedMemory = 0L
这里首先先得到传入的block对应的rdd的id.得到这个rdd_id的目的是淘汰block时,如果发现是这个rdd的block时,不进行淘汰.
val rddToAdd = blockId.flatMap(getRddId)
val selectedBlocks = new ArrayBuffer[BlockId]
// This is synchronized to ensure that the set of entries is not changed
// (because of getValue or getBytes) while traversing the iterator, as that
// can lead to exceptions.
entries.synchronized {
这里开始对storage的内存中所有的cache进行迭代,这个迭代从最先进行cache的block开始,如果迭代到的block对应的RDD不是传入的BLOCK对应的RDD时,把这个BLOCK添加到选择的BLOCK的集合中,并计算当前内存池中的内存是否达到需要的内存空间,如果达到,停止选择BLOCK的操作.
val iterator = entries.entrySet().iterator()
while (freedMemory < space && iterator.hasNext) {
val pair = iterator.next()
val blockId = pair.getKey
if (rddToAdd.isEmpty || rddToAdd != getRddId(blockId)) {
selectedBlocks += blockId
freedMemory += pair.getValue.size
}
}
}
if (freedMemory >= space) {
如果流程执行到这里,说明内存空间释放成功,现在可用的内存空间已经达到需要的内存空间的大小,把选择的BLOCK对应的CACHE通过BLOCKMANAGER从内存中进行释放.并把释放后的BLOCK添加到droppedBlocks的集合中,这个集合用于返回结果,表示这次空间的释放时,这些BLOCK已经从CACHE中称出.
logInfo(s”${selectedBlocks.size} blocks selected for dropping”)
for (blockId <- selectedBlocks) {
val entry = entries.synchronized { entries.get(blockId) }
// This should never be null as only one task should be dropping
// blocks and removing entries. However the check is still here for
// future safety.
if (entry != null) {
val data = if (entry.deserialized) {
Left(entry.value.asInstanceOf[Array[Any]])
} else {
Right(entry.value.asInstanceOf[ByteBuffer].duplicate())
}
val droppedBlockStatus = blockManager.dropFromMemory(blockId, data)
droppedBlockStatus.foreach { status => droppedBlocks += ((blockId,
status)) }
}
}
true
} else {
流程执行到这里,表示STORAGE的内存空间中无法释放出更多的内存,也就相当于是释放空间失败求.
blockId.foreach { id =>
logInfo(s”Will not store $id as it would require dropping another block ” +
“from the same RDD”)
}
false
}
}
}
打印RDD信息 rdd.toDebugString
Spark课程学习地址：
课程介绍：http://edu.51cto.com/course/course_id-1659.html#showDesc
报名参与：http://ke.qq.com/cgi-bin/courseDetail?course_id=6167
Matei Zaharia Spark 论文：http://www.chinacloud.cn/upload/2015-01/15012705072318.pdf
http://zhidao.baidu.com/link?url=b_Gj7HMtBCTi1OU9FBlumNXSC92d57qCdVpUUQlT2F9Qnq6fbk2-OkbhakYpQgv3wag2fb2YeBZTavrkNRg7of7rNVbDzh3iYpIyzEXxScq
http://lxw1234.com/archives/2015/07/350.htm