首页 文章 精选 留言 我的

精选列表

搜索[源码学习]，共10000篇文章
优秀的个人博客,低调大师

Spark源码分析 -- PairRDD

和一般RDD最大的不同就是有两个泛型参数, [K, V]表示pair的概念 关键的function是, combineByKey, 所有pair相关操作的抽象 combine是这样的操作, Turns an RDD[(K, V)] into a result of type RDD[(K, C)] 其中C有可能只是简单类型, 但经常是seq, 比如(Int, Int) to (Int, Seq[Int]) 下面来看看combineByKey的参数, 首先需要用户自定义一些操作, createCombiner: V => C, C不存在的情况下, 比如通过V创建seq C mergeValue: (C, V) => C, 当C已经存在的情况下, 需要merge, 比如把item V加到seq C中, 或者叠加 mergeCombiners: (C, C) => C, 合并两个C partitioner: Partitioner, Shuffle时需要的Partitioner mapSideCombine: Boolean =true, 为了减小传输量, 很多combine可以在map端先做, 比如叠加, 可以先在一个partition中把所有相同的key的value叠加, 再shuffle serializerClass: String =null,传输需要序列化, 用户可以自定义序列化类 /** * Extra functions available on RDDs of (key, value) pairs through an implicit conversion. * Import `org.apache.spark.SparkContext._` at the top of your program to use these functions. */ class PairRDDFunctions[K: ClassManifest, V: ClassManifest](self: RDD[(K, V)]) extends Logging with SparkHadoopMapReduceUtil with Serializable { /** * Generic function to combine the elements for each key using a custom set of aggregation * functions. Turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a "combined type" C * Note that V and C can be different -- for example, one might group an RDD of type * (Int, Int) into an RDD of type (Int, Seq[Int]). Users provide three functions: * * - `createCombiner`, which turns a V into a C (e.g., creates a one-element list) * - `mergeValue`, to merge a V into a C (e.g., adds it to the end of a list) * - `mergeCombiners`, to combine two C's into a single one. * * In addition, users can control the partitioning of the output RDD, and whether to perform * map-side aggregation (if a mapper can produce multiple items with the same key). */ def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializerClass: String = null): RDD[(K, C)] = { val aggregator = new Aggregator[K, V, C](createCombiner, mergeValue, mergeCombiners) //1.Aggregator //RDD本身的partitioner和传入的partitioner相等时, 即不需要重新shuffle, 直接map即可 if (self.partitioner == Some(partitioner)) { self.mapPartitions(aggregator.combineValuesByKey, preservesPartitioning = true) //2. mapPartitions, map端直接调用combineValuesByKey } else if (mapSideCombine) { //如果需要mapSideCombine val combined = self.mapPartitions(aggregator.combineValuesByKey, preservesPartitioning = true) //先在partition内部做mapSideCombine val partitioned = new ShuffledRDD[K, C, (K, C)](combined, partitioner).setSerializer(serializerClass) //3. ShuffledRDD, 进行shuffle partitioned.mapPartitions(aggregator.combineCombinersByKey, preservesPartitioning = true) //Shuffle完后, 在reduce端再做一次combine, 使用combineCombinersByKey } else { // Don't apply map-side combiner.和上面的区别就是不做mapSideCombine // A sanity check to make sure mergeCombiners is not defined. assert(mergeCombiners == null) val values = new ShuffledRDD[K, V, (K, V)](self, partitioner).setSerializer(serializerClass) values.mapPartitions(aggregator.combineValuesByKey, preservesPartitioning = true) } } } 1. Aggregator 在combineByKey中, 首先创建Aggregator, 其实在Aggregator中封装了两个函数, combineValuesByKey, 用于处理将V加入到C的case, 比如加入一个item到一个seq里面, 用于map端 combineCombinersByKey, 用于处理两个C合并, 比如两个seq合并, 用于reduce端 case class Aggregator[K, V, C] ( createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C) { def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]]) : Iterator[(K, C)] = { val combiners = new JHashMap[K, C] for (kv <- iter) { val oldC = combiners.get(kv._1) if (oldC == null) { combiners.put(kv._1, createCombiner(kv._2)) } else { combiners.put(kv._1, mergeValue(oldC, kv._2)) } } combiners.iterator } def combineCombinersByKey(iter: Iterator[(K, C)]) : Iterator[(K, C)] = { val combiners = new JHashMap[K, C] iter.foreach { case(k, c) => val oldC = combiners.get(k) if (oldC == null) { combiners.put(k, c) } else { combiners.put(k, mergeCombiners(oldC, c)) } } combiners.iterator } } 2. mapPartitions mapPartitions其实就是使用MapPartitionsRDD 做的事情就是对当前partition执行map函数f, Iterator[T] => Iterator[U] 比如, 执行combineValuesByKey: Iterator[_ <: Product2[K, V]] to Iterator[(K, C)] /** * Return a new RDD by applying a function to each partition of this RDD. */ def mapPartitions[U: ClassManifest](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U] = new MapPartitionsRDD(this, sc.clean(f), preservesPartitioning) private[spark] class MapPartitionsRDD[U: ClassManifest, T: ClassManifest]( prev: RDD[T], f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false) extends RDD[U](prev) { override val partitioner = if (preservesPartitioning) firstParent[T].partitioner else None override def getPartitions: Array[Partition] = firstParent[T].partitions override def compute(split: Partition, context: TaskContext) = f(firstParent[T].iterator(split, context)) // 对于map,就是调用f 3. ShuffledRDD Shuffle实际上是由系统的shuffleFetcher完成的, Spark的抽象封装非常的好 所以在这里看不到Shuffle具体是怎么样做的, 这个需要分析到shuffleFetcher时候才能看到 因为每个shuffle是有一个全局的shuffleid的 所以在compute里面, 你只是看到用BlockStoreShuffleFetcher根据shuffleid和partitionid直接fetch到shuffle过后的数据 /** * The resulting RDD from a shuffle (e.g. repartitioning of data). * @param prev the parent RDD. * @param part the partitioner used to partition the RDD * @tparam K the key class. * @tparam V the value class. */ class ShuffledRDD[K, V, P <: Product2[K, V] : ClassManifest]( @transient var prev: RDD[P], part: Partitioner) extends RDD[P](prev.context, Nil) { override val partitioner = Some(part) //ShuffleRDD会进行repartition, 所以从Partitioner中取出新的part数目 //并用Array.tabulate动态创建相应个数的ShuffledRDDPartition override def getPartitions: Array[Partition] = { Array.tabulate[Partition](part.numPartitions)(i => new ShuffledRDDPartition(i)) } override def compute(split: Partition, context: TaskContext): Iterator[P] = { val shuffledId = dependencies.head.asInstanceOf[ShuffleDependency[K, V]].shuffleId SparkEnv.get.shuffleFetcher.fetch[P](shuffledId, split.index, context.taskMetrics, SparkEnv.get.serializerManager.get(serializerClass)) } } ShuffledRDDPartition没啥区别, 一样只是记录id private[spark] class ShuffledRDDPartition(val idx: Int) extends Partition { override val index = idx override def hashCode(): Int = idx } 下面再来看一下, 如果使用combineByKey来实现其他的操作的, group group是比较典型的例子, (Int, Int) to (Int, Seq[Int]) 由于groupByKey不使用map side combine, 因为这样也无法减少传输空间, 所以不需要实现mergeCombiners /** * Group the values for each key in the RDD into a single sequence. Allows controlling the * partitioning of the resulting key-value pair RDD by passing a Partitioner. */ def groupByKey(partitioner: Partitioner): RDD[(K, Seq[V])] = { // groupByKey shouldn't use map side combine because map side combine does not // reduce the amount of data shuffled and requires all map side data be inserted // into a hash table, leading to more objects in the old gen. def createCombiner(v: V) = ArrayBuffer(v) //创建seq def mergeValue(buf: ArrayBuffer[V], v: V) = buf += v //添加item到seq val bufs = combineByKey[ArrayBuffer[V]]( createCombiner _, mergeValue _, null, partitioner, mapSideCombine=false) bufs.asInstanceOf[RDD[(K, Seq[V])]] } reduce reduce是更简单的一种情况, 只是两个值合并成一个值, (Int, Int V) to (Int, Int C), 比如叠加 所以createCombiner很简单, 就是直接返回v 而mergeValue和mergeCombiners逻辑是相同的, 没有区别 /** * Merge the values for each key using an associative reduce function. This will also perform * the merging locally on each mapper before sending results to a reducer, similarly to a * "combiner" in MapReduce. */ def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = { combineByKey[V]((v: V) => v, func, func, partitioner) } 本文章摘自博客园，原文发布日期：2013-12-24
2017-05-01
优秀的个人博客,低调大师

Spark 源码分析 -- Task

Task是介于DAGScheduler和TaskScheduler中间的接口 在DAGScheduler, 需要把DAG中的每个stage的每个partitions封装成task 最终把taskset提交给TaskScheduler /** * A task to execute on a worker node. */ private[spark] abstract class Task[T](val stageId: Int) extends Serializable { def run(attemptId: Long): T //Task的核心函数 def preferredLocations: Seq[TaskLocation] = Nil //Spark关注locality,可以选择该task运行的location var epoch: Long = -1 // Map output tracker epoch. Will be set by TaskScheduler. var metrics: Option[TaskMetrics] = None } TaskContext 用于记录TaskMetrics和在Task中用到的callback 比如对于HadoopRDD, task完成时需要close input stream package org.apache.spark class TaskContext( val stageId: Int, val splitId: Int, val attemptId: Long, val runningLocally: Boolean = false, val taskMetrics: TaskMetrics = TaskMetrics.empty() //TaskMetrics封装了task执行时一些指标和数据 ) extends Serializable { @transient val onCompleteCallbacks = new ArrayBuffer[() => Unit] // Add a callback function to be executed on task completion. An example use // is for HadoopRDD to register a callback to close the input stream. def addOnCompleteCallback(f: () => Unit) { onCompleteCallbacks += f } def executeOnCompleteCallbacks() { onCompleteCallbacks.foreach{_()} } } ResultTask 对应于Result Stage直接产生结果 package org.apache.spark.scheduler private[spark] class ResultTask[T, U]( stageId: Int, var rdd: RDD[T], var func: (TaskContext, Iterator[T]) => U, var partition: Int, @transient locs: Seq[TaskLocation], var outputId: Int) extends Task[U](stageId) with Externalizable { override def run(attemptId: Long): U = { // 对于resultTask, run就是返回执行的结果, 比如count值 val context = new TaskContext(stageId, partition, attemptId, runningLocally = false) metrics = Some(context.taskMetrics) try { func(context, rdd.iterator(split, context)) // 直接就是对RDD的iterator调用func, 比如count函数 } finally { context.executeOnCompleteCallbacks() } } } ShuffleMapTask 对应于ShuffleMap Stage, 产生的结果作为其他stage的输入 package org.apache.spark.scheduler private[spark] class ShuffleMapTask( stageId: Int, var rdd: RDD[_], var dep: ShuffleDependency[_,_], var partition: Int, @transient private var locs: Seq[TaskLocation]) extends Task[MapStatus](stageId) with Externalizable with Logging { override def run(attemptId: Long): MapStatus = { val numOutputSplits = dep.partitioner.numPartitions // 从ShuffleDependency的partitioner中获取到shuffle目标partition的个数 val taskContext = new TaskContext(stageId, partition, attemptId, runningLocally = false) metrics = Some(taskContext.taskMetrics) val blockManager = SparkEnv.get.blockManager // shuffle需要借助blockManager来完成 var shuffle: ShuffleBlocks = null var buckets: ShuffleWriterGroup = null try { // Obtain all the block writers for shuffle blocks. val ser = SparkEnv.get.serializerManager.get(dep.serializerClass) shuffle = blockManager.shuffleBlockManager.forShuffle(dep.shuffleId, numOutputSplits, ser) // 创建shuffleBlockManager, 参数是shuffleId和目标partitions数目 buckets = shuffle.acquireWriters(partition) // 生成shuffle目标buckets(对应于partition) // Write the map output to its associated buckets. for (elem <- rdd.iterator(split, taskContext)) { // 从RDD中取出每个elem数据 val pair = elem.asInstanceOf[Product2[Any, Any]] val bucketId = dep.partitioner.getPartition(pair._1) // 根据pair的key进行shuffle, 得到目标bucketid buckets.writers(bucketId).write(pair) // 将pair数据写入bucket } // Commit这些buckets到block, 其他的RDD会从通过shuffleid找到这些block, 并读取数据 // Commit the writes. Get the size of each bucket block (total block size). var totalBytes = 0L val compressedSizes: Array[Byte] = buckets.writers.map { writer: BlockObjectWriter => // 计算所有buckets写入文件data的size总和(压缩值) writer.commit() writer.close() val size = writer.size() totalBytes += size MapOutputTracker.compressSize(size) } // Update shuffle metrics. val shuffleMetrics = new ShuffleWriteMetrics shuffleMetrics.shuffleBytesWritten = totalBytes metrics.get.shuffleWriteMetrics = Some(shuffleMetrics) return new MapStatus(blockManager.blockManagerId, compressedSizes) // 返回值为MapStatus, 包含blockManagerId和写入的data size, 会被注册到MapOutputTracker } catch { case e: Exception => // If there is an exception from running the task, revert the partial writes // and throw the exception upstream to Spark. if (buckets != null) { buckets.writers.foreach(_.revertPartialWrites()) } throw e } finally { // Release the writers back to the shuffle block manager. if (shuffle != null && buckets != null) { shuffle.releaseWriters(buckets) } // Execute the callbacks on task completion. taskContext.executeOnCompleteCallbacks() } } TaskSet 用于封装一个stage的所有的tasks, 以提交给TaskScheduler package org.apache.spark.scheduler /** * A set of tasks submitted together to the low-level TaskScheduler, usually representing * missing partitions of a particular stage. */ private[spark] class TaskSet( val tasks: Array[Task[_]], val stageId: Int, val attempt: Int, val priority: Int, val properties: Properties) { val id: String = stageId + "." + attempt override def toString: String = "TaskSet " + id } 本文章摘自博客园，原文发布日期：2013-12-30
2017-05-01
优秀的个人博客,低调大师

Spark 源码分析 -- Stage

理解stage, 关键就是理解Narrow Dependency和Wide Dependency, 可能还是觉得比较难理解 关键在于是否需要shuffle, 不需要shuffle是可以随意并发的, 所以stage的边界就是需要shuffle的地方, 如下图很清楚 并且Stage分为两种, shuffle map stage, in which case its tasks' results are input for another stage 其实就是,非最终stage, 后面还有其他的stage, 所以它的输出一定是需要shuffle并作为后续的输入result stage, in which case its tasks directly compute the action that initiated a job (e.g. count(), save(), etc) 最终的stage, 没有输出, 而是直接产生结果或存储 1 stage class 这个注释写的很清楚 可以看到stage的RDD参数只有一个RDD, final RDD, 而不是一系列的RDD 因为在一个stage中的所有RDD都是map, partition不会有任何改变, 只是在data依次执行不同的map function 所以对于task scheduler而言, 一个RDD的状况就可以代表这个stage /** * A stage is a set of independent tasks all computing the same function that need to run as part * of a Spark job, where all the tasks have the same shuffle dependencies. Each DAG of tasks run * by the scheduler is split up into stages at the boundaries where shuffle occurs, and then the * DAGScheduler runs these stages in topological order. * * Each Stage can either be a shuffle map stage, in which case its tasks' results are input for * another stage, or a result stage, in which case its tasks directly compute the action that * initiated a job (e.g. count(), save(), etc). For shuffle map stages, we also track the nodes * that each output partition is on. * * Each Stage also has a jobId, identifying the job that first submitted the stage. When FIFO * scheduling is used, this allows Stages from earlier jobs to be computed first or recovered * faster on failure. */ private[spark] class Stage( val id: Int, val rdd: RDD[_], // final RDD val shuffleDep: Option[ShuffleDependency[_,_]], // Output shuffle if stage is a map stage val parents: List[Stage], // 父stage val jobId: Int, callSite: Option[String]) extends Logging { val isShuffleMap = shuffleDep != None // 是否是shuffle map stage, 取决于是否有shuffleDep val numPartitions = rdd.partitions.size val outputLocs = Array.fill[List[MapStatus]](numPartitions)(Nil) // 用于buffer每个shuffle中每个maptask的MapStatus var numAvailableOutputs = 0 private var nextAttemptId = 0 def isAvailable: Boolean = { if (!isShuffleMap) { true } else { numAvailableOutputs == numPartitions } } } 2 newStage 如果是shuffle map stage, 需要在这里向mapOutputTracker注册shuffle /** * Create a Stage for the given RDD, either as a shuffle map stage (for a ShuffleDependency) or * as a result stage for the final RDD used directly in an action. The stage will also be * associated with the provided jobId. */ private def newStage( rdd: RDD[_], shuffleDep: Option[ShuffleDependency[_,_]], jobId: Int, callSite: Option[String] = None) : Stage = { if (shuffleDep != None) { // Kind of ugly: need to register RDDs with the cache and map output tracker here // since we can't do it in the RDD constructor because # of partitions is unknown logInfo("Registering RDD " + rdd.id + " (" + rdd.origin + ")") mapOutputTracker.registerShuffle(shuffleDep.get.shuffleId, rdd.partitions.size) } val id = nextStageId.getAndIncrement() val stage = new Stage(id, rdd, shuffleDep, getParentStages(rdd, jobId), jobId, callSite) stageIdToStage(id) = stage stageToInfos(stage) = StageInfo(stage) stage } 3 getMissingParentStages 可以根据final stage的deps找出所有的parent stage private def getMissingParentStages(stage: Stage): List[Stage] = { val missing = new HashSet[Stage] val visited = new HashSet[RDD[_]] def visit(rdd: RDD[_]) { if (!visited(rdd)) { visited += rdd if (getCacheLocs(rdd).contains(Nil)) { for (dep <- rdd.dependencies) { dep match { case shufDep: ShuffleDependency[_,_] => // 如果发现ShuffleDependency, 说明遇到新的stage val mapStage = getShuffleMapStage(shufDep, stage.jobId) // check shuffleToMapStage, 如果该stage已经被创建则直接返回, 否则newStage if (!mapStage.isAvailable) { missing += mapStage } case narrowDep: NarrowDependency[_] => // 对于NarrowDependency, 说明仍然在这个stage中 visit(narrowDep.rdd) } } } } } visit(stage.rdd) missing.toList } 本文章摘自博客园，原文发布日期：2013-12-26
2017-05-01
优秀的个人博客,低调大师

Spark源码分析 -- SchedulableBuilder

SchedulableBuilder就是对Scheduleable tree的封装, 在Pool层面(中间节点), 完成对TaskSet的调度(FIFO, FAIR) 在TaskSetManager 层面(叶子节点), 完成对TaskSet中task的调度(locality)以及track(retry) TaskSetManager 用于封装TaskSet, 主要提供对单个TaskSet内部的tasks的track和schedule 所以主要的接口, resourceOffer, 对于一个resource offer, 如何schedule一个task来执行 statusUpdate, 对于task状态的track /** * Tracks and schedules the tasks within a single TaskSet. This class keeps track of the status of * each task and is responsible for retries on failure and locality. The main interfaces to it * are resourceOffer, which asks the TaskSet whether it wants to run a task on one node, and * statusUpdate, which tells it that one of its tasks changed state (e.g. finished). * * THREADING: This class is designed to only be called from code with a lock on the TaskScheduler * (e.g. its event handlers). It should not be called from other threads. */ private[spark] trait TaskSetManager extends Schedulable { def schedulableQueue = null def schedulingMode = SchedulingMode.NONE def taskSet: TaskSet def resourceOffer( execId: String, host: String, availableCpus: Int, maxLocality: TaskLocality.TaskLocality) : Option[TaskDescription] def statusUpdate(tid: Long, state: TaskState, serializedData: ByteBuffer) def error(message: String) } ClusterTaskSetManager ClusterScheduler上对于TaskSetManager的实现 1 addPendingTasklocality, 在schedule时候需要考虑, 应该优先执行尽可能近的task 所有未被执行的tasks, 都是pending task, 并且是安装不同locality粒度存储在hashmap中的 pendingTasksForExecutor, hashmap, 每个executor被指定的task pendingTasksForHost, hashmap, 每个instance被指定的task pendingTasksForRack, hashmap, 每个机架被指定的task pendingTasksWithNoPrefs, ArrayBuffer, 没有locality preferences的tasks, 随便在那边执行 allPendingTasks, ArrayBuffer, 所有的pending task speculatableTasks, 重复的task, 熟悉hadoop的应该容易理解 可以继续看下addPendingTask, 如何把task加到各个list上去 addPendingTask(index: Int, readding: Boolean =false) 两个参数, index, task的index, 用于从taskset中取得task readding, 表示是否新的task, 因为当executor失败的时候, 也需要把task重新再加到各个list中, list中有重复的task是没有关系的, 因为选取task的时候会自动忽略已经run的task 2 resourceOffer解决如何在taskset内部schedule一个task, 主要需要考虑的是locality, 直接看注释 其中比较意思的是, 对currentLocalityIndex的维护 初始时为0, PROCESS_LOCAL, 只能选择PendingTasksForExecutor 每次调用resourceOffer, 都会计算和前一次task launch之间的时间间隔, 如果超时(各个locality的超时时间不同), currentLocalityIndex会加1, 即不断的放宽 而代表前一次的lastLaunchTime, 只有在resourceOffer中成功的findTask时会被更新, 所以逻辑就是优先选择更local的task, 但当findTask总失败时, 说明需要放宽 但是放宽后, 当有比较local的task被选中时, 这个currentLocalityIndex还会缩小, 因为每次都会把tasklocality赋值给currentLocality 3 statusUpdate应对statusUpdate, 主要是通过在clusterScheduler中注册的listener通知DAGScheduler 当然对于失败的task, 还要再加到pending list里面去 /** * Schedules the tasks within a single TaskSet in the ClusterScheduler. This class keeps track of * the status of each task, retries tasks if they fail (up to a limited number of times), and * handles locality-aware scheduling for this TaskSet via delay scheduling. The main interfaces * to it are resourceOffer, which asks the TaskSet whether it wants to run a task on one node, * and statusUpdate, which tells it that one of its tasks changed state (e.g. finished). * * THREADING: This class is designed to only be called from code with a lock on the * ClusterScheduler (e.g. its event handlers). It should not be called from other threads. */ private[spark] class ClusterTaskSetManager( sched: ClusterScheduler, val taskSet: TaskSet, clock: Clock = SystemClock) extends TaskSetManager with Logging { val tasks = taskSet.tasks val numTasks = tasks.length // Set of pending tasks for each executor. These collections are actually // treated as stacks, in which new tasks are added to the end of the // ArrayBuffer and removed from the end. This makes it faster to detect // tasks that repeatedly fail because whenever a task failed, it is put // back at the head of the stack. They are also only cleaned up lazily; // when a task is launched, it remains in all the pending lists except // the one that it was launched from, but gets removed from them later. private val pendingTasksForExecutor = new HashMap[String, ArrayBuffer[Int]] // Set of pending tasks for each host. Similar to pendingTasksForExecutor, // but at host level. private val pendingTasksForHost = new HashMap[String, ArrayBuffer[Int]] // Set of pending tasks for each rack -- similar to the above. private val pendingTasksForRack = new HashMap[String, ArrayBuffer[Int]] // Set containing pending tasks with no locality preferences. val pendingTasksWithNoPrefs = new ArrayBuffer[Int] // Set containing all pending tasks (also used as a stack, as above). val allPendingTasks = new ArrayBuffer[Int] // Tasks that can be speculated. Since these will be a small fraction of total // tasks, we'll just hold them in a HashSet. val speculatableTasks = new HashSet[Int] // Figure out which locality levels we have in our TaskSet, so we can do delay scheduling val myLocalityLevels = computeValidLocalityLevels() // 当前TaskSet里面的task locality有哪些 val localityWaits = myLocalityLevels.map(getLocalityWait) // 每个locality level默认的等待时间(从配置读) // Delay scheduling variables: we keep track of our current locality level and the time we // last launched a task at that level, and move up a level when localityWaits[curLevel] expires. // We then move down if we manage to launch a "more local" task. var currentLocalityIndex = 0 // 当前myLocalityLevels中的index, 从0开始, 从最小的开始schedule var lastLaunchTime = clock.getTime() // 记录最后launch task的时间, 用于后面会算超时, 如果发生超时, currentLocalityIndex+1 /** * Add a task to all the pending-task lists that it should be on. If readding is set, we are * re-adding the task so only include it in each list if it's not already there. */ private def addPendingTask(index: Int, readding: Boolean = false) { // Utility method that adds `index` to a list only if readding=false or it's not already there def addTo(list: ArrayBuffer[Int]) { if (!readding || !list.contains(index)) { // 新的的task或在该list里面没有 list += index } } var hadAliveLocations = false for (loc <- tasks(index).preferredLocations) { for (execId <- loc.executorId) { if (sched.isExecutorAlive(execId)) { addTo(pendingTasksForExecutor.getOrElseUpdate(execId, new ArrayBuffer)) // 首先加到相应的executor列表中 hadAliveLocations = true } } if (sched.hasExecutorsAliveOnHost(loc.host)) { addTo(pendingTasksForHost.getOrElseUpdate(loc.host, new ArrayBuffer)) // 加到host的列表中 for (rack <- sched.getRackForHost(loc.host)) { addTo(pendingTasksForRack.getOrElseUpdate(rack, new ArrayBuffer)) // 加到Rack的列表中 } hadAliveLocations = true } } if (!hadAliveLocations) { // 如果上面的选择都失败了, 或本来就没有preferred locations, 那就加到pendingTasksWithNoPrefs中 // Even though the task might've had preferred locations, all of those hosts or executors // are dead; put it in the no-prefs list so we can schedule it elsewhere right away. addTo(pendingTasksWithNoPrefs) } if (!readding) { // 对于新的task, 需要加到allPendingTasks中 allPendingTasks += index // No point scanning this whole list to find the old task there } } /** * Dequeue a pending task for a given node and return its index and locality level. * Only search for tasks matching the given locality constraint. */ private def findTask(execId: String, host: String, locality: TaskLocality.Value) : Option[(Int, TaskLocality.Value)] = { // 先从Executor for (index <- findTaskFromList(getPendingTasksForExecutor(execId))) { // findTaskFromList, Dequeue a pending task from the given list and return its index. return Some((index, TaskLocality.PROCESS_LOCAL)) } // Node, 需要先check locality if (TaskLocality.isAllowed(locality, TaskLocality.NODE_LOCAL)) { // locality >= TaskLocality.NODE_LOCAL for (index <- findTaskFromList(getPendingTasksForHost(host))) { return Some((index, TaskLocality.NODE_LOCAL)) } } // Rack, 需要先check locality if (TaskLocality.isAllowed(locality, TaskLocality.RACK_LOCAL)) { for { rack <- sched.getRackForHost(host) index <- findTaskFromList(getPendingTasksForRack(rack)) } { return Some((index, TaskLocality.RACK_LOCAL)) } } // Look for no-pref tasks after rack-local tasks since they can run anywhere. for (index <- findTaskFromList(pendingTasksWithNoPrefs)) { return Some((index, TaskLocality.PROCESS_LOCAL)) } if (TaskLocality.isAllowed(locality, TaskLocality.ANY)) { for (index <- findTaskFromList(allPendingTasks)) { return Some((index, TaskLocality.ANY)) } } // Finally, if all else has failed, find a speculative task return findSpeculativeTask(execId, host, locality) } /** * Respond to an offer of a single executor from the scheduler by finding a task */ override def resourceOffer( execId: String, host: String, availableCpus: Int, maxLocality: TaskLocality.TaskLocality) : Option[TaskDescription] = { if (tasksFinished < numTasks && availableCpus >= CPUS_PER_TASK) { // 前提是task没有执行完和有足够的available cores(>1) val curTime = clock.getTime() var allowedLocality = getAllowedLocalityLevel(curTime) // 取到当前allowed LocalityLevel if (allowedLocality > maxLocality) { // 不能超出作为参数传入的maxLocality, 调用者限定 allowedLocality = maxLocality // We're not allowed to search for farther-away tasks } findTask(execId, host, allowedLocality) match { // 调用findTask, 并对返回值进行case, findTask逻辑很简单就是依次从不同的locality中取task case Some((index, taskLocality)) => { // Found a task; do some bookkeeping and return a task description val task = tasks(index) val taskId = sched.newTaskId() // Figure out whether this should count as a preferred launch logInfo("Starting task %s:%d as TID %s on executor %s: %s (%s)".format( taskSet.id, index, taskId, execId, host, taskLocality)) // Do various bookkeeping copiesRunning(index) += 1 val info = new TaskInfo(taskId, index, curTime, execId, host, taskLocality) taskInfos(taskId) = info taskAttempts(index) = info :: taskAttempts(index) // Update our locality level for delay scheduling currentLocalityIndex = getLocalityIndex(taskLocality) // 用当前Task的locality来更新currentLocalityIndex, 这里index有可能会减少, 因为taskLocality <= currentLocality lastLaunchTime = curTime // 更新lastLaunchTime // Serialize and return the task val startTime = clock.getTime() // We rely on the DAGScheduler to catch non-serializable closures and RDDs, so in here // we assume the task can be serialized without exceptions. val serializedTask = Task.serializeWithDependencies( task, sched.sc.addedFiles, sched.sc.addedJars, ser) val timeTaken = clock.getTime() - startTime increaseRunningTasks(1) logInfo("Serialized task %s:%d as %d bytes in %d ms".format( taskSet.id, index, serializedTask.limit, timeTaken)) val taskName = "task %s:%d".format(taskSet.id, index) if (taskAttempts(index).size == 1) taskStarted(task,info) return Some(new TaskDescription(taskId, execId, taskName, index, serializedTask)) // 最终返回schedule得到的那个task } case _ => } } return None } /** * Get the level we can launch tasks according to delay scheduling, based on current wait time. */ private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = { while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) && // 发生超时 currentLocalityIndex < myLocalityLevels.length - 1) { // Jump to the next locality level, and remove our waiting time for the current one since // we don't want to count it again on the next one lastLaunchTime += localityWaits(currentLocalityIndex) currentLocalityIndex += 1 // currentLocalityIndex 加 1 } myLocalityLevels(currentLocalityIndex) } /** * Find the index in myLocalityLevels for a given locality. This is also designed to work with * localities that are not in myLocalityLevels (in case we somehow get those) by returning the * next-biggest level we have. Uses the fact that the last value in myLocalityLevels is ANY. */ def getLocalityIndex(locality: TaskLocality.TaskLocality): Int = { // 查询locality在myLocalityLevels中的index var index = 0 while (locality > myLocalityLevels(index)) { index += 1 } index } /** * Compute the locality levels used in this TaskSet. Assumes that all tasks have already been * added to queues using addPendingTask. */ // 仅仅从各个pending list中看看当前的taskset中的task有哪些preference locality, 从小到大 private def computeValidLocalityLevels(): Array[TaskLocality.TaskLocality] = { import TaskLocality.{PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL, ANY} val levels = new ArrayBuffer[TaskLocality.TaskLocality] if (!pendingTasksForExecutor.isEmpty && getLocalityWait(PROCESS_LOCAL) != 0) { levels += PROCESS_LOCAL } if (!pendingTasksForHost.isEmpty && getLocalityWait(NODE_LOCAL) != 0) { levels += NODE_LOCAL } if (!pendingTasksForRack.isEmpty && getLocalityWait(RACK_LOCAL) != 0) { levels += RACK_LOCAL } levels += ANY logDebug("Valid locality levels for " + taskSet + ": " + levels.mkString(", ")) levels.toArray } /** Called by cluster scheduler when one of our tasks changes state */ override def statusUpdate(tid: Long, state: TaskState, serializedData: ByteBuffer) { SparkEnv.set(env) state match { case TaskState.FINISHED => taskFinished(tid, state, serializedData) case TaskState.LOST => taskLost(tid, state, serializedData) case TaskState.FAILED => taskLost(tid, state, serializedData) case TaskState.KILLED => taskLost(tid, state, serializedData) case _ => } } def taskStarted(task: Task[_], info: TaskInfo) { sched.listener.taskStarted(task, info) } } Pool 一种对schedulableQueue的抽象, 什么是schedulable? 注释说的, 包含Pools and TaskSetManagers, 这里设计有问题, 你会发现Pools和TaskSetManagers的核心接口完全不同, 虽然TaskSetManagers里面也实现了这些接口, 但都是meanless的 简单理解成, 作者想要统一对待, 泛化Pools和TaskSetManagers, 所以这样做了 所以对于Pool, 可以理解为TaskSetManagers的容器, 当然由于Pool本身也是Schedulable, 所以容器里面也可以放Pool 核心接口getSortedTaskSetQueue, 通过配置不同的SchedulingAlgorithm来调度TaskSetManagers(或pool) 所以注意那些FIFO或FAIR都是用来调度TaskSet的, 所以Spark调度的基础是stage /** * An interface for schedulable entities. * there are two type of Schedulable entities(Pools and TaskSetManagers) */ private[spark] trait Schedulable { var parent: Schedulable // child queues def schedulableQueue: ArrayBuffer[Schedulable] def schedulingMode: SchedulingMode def weight: Int def minShare: Int def runningTasks: Int def priority: Int def stageId: Int def name: String def increaseRunningTasks(taskNum: Int): Unit def decreaseRunningTasks(taskNum: Int): Unit def addSchedulable(schedulable: Schedulable): Unit def removeSchedulable(schedulable: Schedulable): Unit def getSchedulableByName(name: String): Schedulable def executorLost(executorId: String, host: String): Unit def checkSpeculatableTasks(): Boolean def getSortedTaskSetQueue(): ArrayBuffer[TaskSetManager] def hasPendingTasks(): Boolean } package org.apache.spark.scheduler.cluster //An Schedulable entity that represent collection of Pools or TaskSetManagers private[spark] class Pool( val poolName: String, val schedulingMode: SchedulingMode, initMinShare: Int, initWeight: Int) extends Schedulable with Logging { var schedulableQueue = new ArrayBuffer[Schedulable] // 用于buffer Schedulable, TaskSetManager var schedulableNameToSchedulable = new HashMap[String, Schedulable] var priority = 0 var stageId = 0 var name = poolName var parent:Schedulable = null var taskSetSchedulingAlgorithm: SchedulingAlgorithm = { // SchedulingAlgorithm其实就是定义comparator,后面好将TaskSet排序 schedulingMode match { case SchedulingMode.FAIR => new FairSchedulingAlgorithm() // Fair case SchedulingMode.FIFO => new FIFOSchedulingAlgorithm() // FIFO } } override def addSchedulable(schedulable: Schedulable) { // 增加一个TaskSetManager schedulableQueue += schedulable schedulableNameToSchedulable(schedulable.name) = schedulable schedulable.parent= this } override def removeSchedulable(schedulable: Schedulable) { // 删除一个TaskSetManager schedulableQueue -= schedulable schedulableNameToSchedulable -= schedulable.name } override def getSortedTaskSetQueue(): ArrayBuffer[TaskSetManager] = { // 返回排过序的TaskSetManager列表 var sortedTaskSetQueue = new ArrayBuffer[TaskSetManager] val sortedSchedulableQueue = schedulableQueue.sortWith(taskSetSchedulingAlgorithm.comparator) // sortWith for (schedulable <- sortedSchedulableQueue) { sortedTaskSetQueue ++= schedulable.getSortedTaskSetQueue() // 这里的schedulable有可能也是pool, 所以需要递归调用 } return sortedTaskSetQueue } } SchedulableBuilder 上面说了Pool里面可以是TaskSetManagers也可以是pool, 这样是不是可以形成tree SchedulableBuilder就是对Schedulable Tree的封装, 通过TaskSetManagers(叶节点)和pools(中间节点), 来生成Schedulable Tree 这里只列出最简单的FIFO, 看不出tree的感觉 对于FIFO很简单, 直接使用一个Pool就可以, 把所有的TaskSet使用addSchedulable加进去, 然后排序读出来即可 这里没有列出Fair的实现, 比较复杂, 后面再分析吧 /** * An interface to build Schedulable tree * buildPools: build the tree nodes(pools) * addTaskSetManager: build the leaf nodes(TaskSetManagers) */ private[spark] trait SchedulableBuilder { def buildPools() def addTaskSetManager(manager: Schedulable, properties: Properties) } private[spark] class FIFOSchedulableBuilder(val rootPool: Pool) extends SchedulableBuilder with Logging { override def buildPools() { // nothing } override def addTaskSetManager(manager: Schedulable, properties: Properties) { rootPool.addSchedulable(manager) } } 本文章摘自博客园，原文发布日期：2014-01-06
2017-05-01
优秀的个人博客,低调大师

Spark 源码分析 -- BlockStore

BlockStore 抽象接口类, 关键get和put都有两个版本 序列化, putBytes, getBytes 非序列化, putValues, getValues 其中putValues的返回值为PutResult, 其中的data可能是Iterator或ByteBuffer private[spark] case class PutResult(size: Long, data: Either[Iterator[_], ByteBuffer]) /** * Abstract class to store blocks */ private[spark] abstract class BlockStore(val blockManager: BlockManager) extends Logging { def putBytes(blockId: String, bytes: ByteBuffer, level: StorageLevel) /** * Put in a block and, possibly, also return its content as either bytes or another Iterator. * This is used to efficiently write the values to multiple locations (e.g. for replication). * * @return a PutResult that contains the size of the data, as well as the values put if * returnValues is true (if not, the result's data field can be null) */ def putValues(blockId: String, values: ArrayBuffer[Any], level: StorageLevel, returnValues: Boolean) : PutResult /** * Return the size of a block in bytes. */ def getSize(blockId: String): Long def getBytes(blockId: String): Option[ByteBuffer] def getValues(blockId: String): Option[Iterator[Any]] /** * Remove a block, if it exists. * @param blockId the block to remove. * @return True if the block was found and removed, False otherwise. */ def remove(blockId: String): Boolean def contains(blockId: String): Boolean def clear() { } } DiskStore 对应DiskStore其实很单纯, 就是打开相应的文件读或写. /** * Stores BlockManager blocks on disk. */ private class DiskStore(blockManager: BlockManager, rootDirs: String) extends BlockStore(blockManager) with Logging { override def putBytes(blockId: String, _bytes: ByteBuffer, level: StorageLevel) { // So that we do not modify the input offsets ! // duplicate does not copy buffer, so inexpensive val bytes = _bytes.duplicate() val file = createFile(blockId) val channel = new RandomAccessFile(file, "rw").getChannel() while (bytes.remaining > 0) { channel.write(bytes) } channel.close() } override def putValues( blockId: String, values: ArrayBuffer[Any], level: StorageLevel, returnValues: Boolean) : PutResult = { val file = createFile(blockId) val fileOut = blockManager.wrapForCompression(blockId, new FastBufferedOutputStream(new FileOutputStream(file))) val objOut = blockManager.defaultSerializer.newInstance().serializeStream(fileOut) objOut.writeAll(values.iterator) objOut.close() val length = file.length() if (returnValues) { // Return a byte buffer for the contents of the file val buffer = getFileBytes(file) PutResult(length, Right(buffer)) } else { PutResult(length, null) } } override def getBytes(blockId: String): Option[ByteBuffer] = { val file = getFile(blockId) val bytes = getFileBytes(file) Some(bytes) } override def getValues(blockId: String): Option[Iterator[Any]] = { getBytes(blockId).map(bytes => blockManager.dataDeserialize(blockId, bytes)) } } MemoryStore 对于MemoryStore复杂一些 首先使用LinkedHashMap, 可遍历的HashMap, 来组织MemoryStore, 其中的hashmap的结构(blockid, entry) 使用Entry抽象来表示block内容 并且在put的时候, 还涉及到memory空间的释放, ensureFreeSpace /** * Stores blocks in memory, either as ArrayBuffers of deserialized Java objects or as * serialized ByteBuffers. */ private class MemoryStore(blockManager: BlockManager, maxMemory: Long) extends BlockStore(blockManager) { // 使用Entry来表示block内容 case class Entry(value: Any, size: Long, deserialized: Boolean, var dropPending: Boolean = false) private val entries = new LinkedHashMap[String, Entry](32, 0.75f, true) // 使用LinkedHashMap来表示整个MemoryStore private var currentMemory = 0L // Object used to ensure that only one thread is putting blocks and if necessary, dropping // blocks from the memory store. private val putLock = new Object() // HashMap不是线程安全的, 需要锁同步 override def putBytes(blockId: String, _bytes: ByteBuffer, level: StorageLevel) { // Work on a duplicate - since the original input might be used elsewhere. val bytes = _bytes.duplicate() bytes.rewind() // 对于NIO的ByteBuffer, 使用前最好rewind if (level.deserialized) { // 如果storage level需要非序列化的 val values = blockManager.dataDeserialize(blockId, bytes) // 需要先反序列化 val elements = new ArrayBuffer[Any] elements ++= values val sizeEstimate = SizeEstimator.estimate(elements.asInstanceOf[AnyRef]) tryToPut(blockId, elements, sizeEstimate, true) } else { tryToPut(blockId, bytes, bytes.limit, false) } } // putValues的返回值取决于storage level, 如果是deserialized, 返回iterator, 否则ByteBuffer override def putValues( blockId: String, values: ArrayBuffer[Any], level: StorageLevel, returnValues: Boolean) : PutResult = { if (level.deserialized) { val sizeEstimate = SizeEstimator.estimate(values.asInstanceOf[AnyRef]) tryToPut(blockId, values, sizeEstimate, true) PutResult(sizeEstimate, Left(values.iterator)) } else { val bytes = blockManager.dataSerialize(blockId, values.iterator) tryToPut(blockId, bytes, bytes.limit, false) PutResult(bytes.limit(), Right(bytes.duplicate())) } } override def getBytes(blockId: String): Option[ByteBuffer] = { val entry = entries.synchronized { entries.get(blockId) } if (entry == null) { None } else if (entry.deserialized) { Some(blockManager.dataSerialize(blockId, entry.value.asInstanceOf[ArrayBuffer[Any]].iterator)) } else { Some(entry.value.asInstanceOf[ByteBuffer].duplicate()) // Doesn't actually copy the data } } override def getValues(blockId: String): Option[Iterator[Any]] = { val entry = entries.synchronized { entries.get(blockId) } if (entry == null) { None } else if (entry.deserialized) { Some(entry.value.asInstanceOf[ArrayBuffer[Any]].iterator) } else { val buffer = entry.value.asInstanceOf[ByteBuffer].duplicate() // Doesn't actually copy data Some(blockManager.dataDeserialize(blockId, buffer)) } } /** * Try to put in a set of values, if we can free up enough space. The value should either be * an ArrayBuffer if deserialized is true or a ByteBuffer otherwise. Its (possibly estimated) * size must also be passed by the caller. * * Locks on the object putLock to ensure that all the put requests and its associated block * dropping is done by only on thread at a time. Otherwise while one thread is dropping * blocks to free memory for one block, another thread may use up the freed space for * another block. */ private def tryToPut(blockId: String, value: Any, size: Long, deserialized: Boolean): Boolean = { // TODO: Its possible to optimize the locking by locking entries only when selecting blocks // to be dropped. Once the to-be-dropped blocks have been selected, and lock on entries has been // released, it must be ensured that those to-be-dropped blocks are not double counted for // freeing up more space for another block that needs to be put. Only then the actually dropping // of blocks (and writing to disk if necessary) can proceed in parallel. putLock.synchronized { if (ensureFreeSpace(blockId, size)) { // 如果可用分配足够的memory val entry = new Entry(value, size, deserialized) entries.synchronized { entries.put(blockId, entry) } currentMemory += size true } else { // 如果memory无法放下这个block, 那么只有从memory删除, 如果可以用disk, 那么在dropFromMemory中会put到disk中 // Tell the block manager that we couldn't put it in memory so that it can drop it to // disk if the block allows disk storage. val data = if (deserialized) { Left(value.asInstanceOf[ArrayBuffer[Any]]) } else { Right(value.asInstanceOf[ByteBuffer].duplicate()) } blockManager.dropFromMemory(blockId, data) false } } } /** * Tries to free up a given amount of space to store a particular block, but can fail and return * false if either the block is bigger than our memory or it would require replacing another * block from the same RDD (which leads to a wasteful cyclic replacement pattern for RDDs that * don't fit into memory that we want to avoid). * * Assumes that a lock is held by the caller to ensure only one thread is dropping blocks. * Otherwise, the freed space may fill up before the caller puts in their new value. */ private def ensureFreeSpace(blockIdToAdd: String, space: Long): Boolean = { if (space > maxMemory) { logInfo("Will not store " + blockIdToAdd + " as it is larger than our memory limit") return false } if (maxMemory - currentMemory < space) { val rddToAdd = getRddId(blockIdToAdd) val selectedBlocks = new ArrayBuffer[String]() var selectedMemory = 0L // 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() // 会依次删除现有的block, 直到可以放下新的block while (maxMemory - (currentMemory - selectedMemory) < space && iterator.hasNext) { val pair = iterator.next() val blockId = pair.getKey if (rddToAdd != null && rddToAdd == getRddId(blockId)) { logInfo("Will not store " + blockIdToAdd + " as it would require dropping another " + "block from the same RDD") return false } selectedBlocks += blockId selectedMemory += pair.getValue.size } } if (maxMemory - (currentMemory - selectedMemory) >= space) { logInfo(selectedBlocks.size + " blocks selected for dropping") for (blockId <- selectedBlocks) { // 删除selectedBlocks, 释放空间 val entry = entries.synchronized { entries.get(blockId) } // This should never be null as only one thread 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[ArrayBuffer[Any]]) } else { Right(entry.value.asInstanceOf[ByteBuffer].duplicate()) } blockManager.dropFromMemory(blockId, data) } } return true } else { return false } } return true } 本文章摘自博客园，原文发布日期：2014-01-09
2017-05-01
优秀的个人博客,低调大师

Spark源码分析 – Executor

ExecutorBackend 很简单的接口 package org.apache.spark.executor /** * A pluggable interface used by the Executor to send updates to the cluster scheduler. */ private[spark] trait ExecutorBackend { def statusUpdate(taskId: Long, state: TaskState, data: ByteBuffer) } StandaloneExecutorBackend 维护executor, 并负责注册executor以及executor和driver之间的通信 private[spark] class StandaloneExecutorBackend( driverUrl: String, executorId: String, hostPort: String, cores: Int) extends Actor with ExecutorBackend with Logging { var executor: Executor = null var driver: ActorRef = null override def preStart() { logInfo("Connecting to driver: " + driverUrl) driver = context.actorFor(driverUrl) // 创建driver actor ref, 以便于和driver通信 driver ! RegisterExecutor(executorId, hostPort, cores) // 向driver注册executor } override def receive = { case RegisteredExecutor(sparkProperties) => logInfo("Successfully registered with driver") // Make this host instead of hostPort ? executor = new Executor(executorId, Utils.parseHostPort(hostPort)._1, sparkProperties) // 当注册成功后, 创建Executor case RegisterExecutorFailed(message) => logError("Slave registration failed: " + message) System.exit(1) case LaunchTask(taskDesc) => logInfo("Got assigned task " + taskDesc.taskId) if (executor == null) { logError("Received launchTask but executor was null") System.exit(1) } else { executor.launchTask(this, taskDesc.taskId, taskDesc.serializedTask) // 调用executor.launchTask,启动task } case Terminated(_) | RemoteClientDisconnected(_, _) | RemoteClientShutdown(_, _) => logError("Driver terminated or disconnected! Shutting down.") System.exit(1) } override def statusUpdate(taskId: Long, state: TaskState, data: ByteBuffer) { driver ! StatusUpdate(executorId, taskId, state, data) // 当task状态变化时, 报告给driver actor } } Executor 对于Executor, 维护一个threadPool, 可以run多个task, 取决于core的个数 所以对于launchTask, 就是在threadPool中挑个thread去run TaskRunner private[spark] class Executor( executorId: String, slaveHostname: String, properties: Seq[(String, String)]) extends Logging { // Initialize Spark environment (using system properties read above) val env = SparkEnv.createFromSystemProperties(executorId, slaveHostname, 0, false, false) SparkEnv.set(env) // Start worker thread pool val threadPool = new ThreadPoolExecutor( 1, 128, 600, TimeUnit.SECONDS, new SynchronousQueue[Runnable]) def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) { threadPool.execute(new TaskRunner(context, taskId, serializedTask)) } } TaskRunner class TaskRunner(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) extends Runnable { override def run() { try { SparkEnv.set(env) Accumulators.clear() val (taskFiles, taskJars, taskBytes) = Task.deserializeWithDependencies(serializedTask) // 反序列化 updateDependencies(taskFiles, taskJars) val task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.getContextClassLoader) // 反序列化 attemptedTask = Some(task) logInfo("Its epoch is " + task.epoch) env.mapOutputTracker.updateEpoch(task.epoch) taskStart = System.currentTimeMillis() val value = task.run(taskId.toInt) // 调用task.run执行真正的逻辑 val taskFinish = System.currentTimeMillis() val accumUpdates = Accumulators.values val result = new TaskResult(value, accumUpdates, task.metrics.getOrElse(null)) // 生成TaskResult val serializedResult = ser.serialize(result) // 将TaskResult序列化 logInfo("Serialized size of result for " + taskId + " is " + serializedResult.limit) context.statusUpdate(taskId, TaskState.FINISHED, serializedResult) // 将任务完成和taskresult,通过statusUpdate报告给driver logInfo("Finished task ID " + taskId) } catch { // 处理各种fail, 同样也要用statusUpdate event通知driver case ffe: FetchFailedException => { val reason = ffe.toTaskEndReason context.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason)) } case t: Throwable => { val serviceTime = (System.currentTimeMillis() - taskStart).toInt val metrics = attemptedTask.flatMap(t => t.metrics) for (m <- metrics) { m.executorRunTime = serviceTime m.jvmGCTime = getTotalGCTime - startGCTime } val reason = ExceptionFailure(t.getClass.getName, t.toString, t.getStackTrace, metrics) context.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason)) // TODO: Should we exit the whole executor here? On the one hand, the failed task may // have left some weird state around depending on when the exception was thrown, but on // the other hand, maybe we could detect that when future tasks fail and exit then. logError("Exception in task ID " + taskId, t) //System.exit(1) } } } } 本文章摘自博客园，原文发布日期：2014-01-07
2017-05-01
优秀的个人博客,低调大师

Spark源码分析 – Checkpoint

CP的步骤 1. 首先如果RDD需要CP, 调用RDD.checkpoint()来mark 注释说了, 这个需要在Job被执行前被mark, 原因后面看, 并且最好选择persist这个RDD, 否则在存CP文件时需要重新computeRDD内容 并且当RDD被CP后, 所有dependencies都会被清除, 因为既然RDD已经被CP, 那么就可以直接从文件读取, 没有必要保留之前的parents的dependencies(保留只是为了replay) 2. 在SparkContext.runJob中, 最后会调用rdd.doCheckpoint() 如果前面已经mark过, 那么这里就会将rdd真正CP到文件中去, 这也是前面为什么说, mark必须在run job之前完成 def runJob[T, U: ClassManifest]( rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], allowLocal: Boolean, resultHandler: (Int, U) => Unit) { val result = dagScheduler.runJob(rdd, func, partitions, callSite, allowLocal, resultHandler, localProperties.get) rdd.doCheckpoint() result } 3. 在RDDCheckpointData.doCheckpoint中 会调用rdd.markCheckpointed(newRDD), 清除dependencies信息 并最终将状态设为, Checkpointed, 以表示完成CP 4. Checkpoint如何使用, 在RDD.computeOrReadCheckpoint中, 看到如果已经完成CP, 会直接从firstParent中读数据, 刚开始会觉得比较奇怪 if (isCheckpointed) { firstParent[T].iterator(split, context) } RDD.firstParent的定义如下, 就是从dependencies中取第一个dependency的rdd dependencies.head.rdd.asInstanceOf[RDD[U]] 而RDD.dependencies的定义如下, 可用看到在完成CP的情况下, 从dependencies中读到的其实就是CP RDD, 所以可以直接用 checkpointRDD.map(r => List(new OneToOneDependency(r))).getOrElse {dependencies_} RDD /** An Option holding our checkpoint RDD, if we are checkpointed */ private def checkpointRDD: Option[RDD[T]] = checkpointData.flatMap(_.checkpointRDD) private[spark] var checkpointData: Option[RDDCheckpointData[T]] = None // Avoid handling doCheckpoint multiple times to prevent excessive recursion private var doCheckpointCalled = false /** * 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() { if (context.checkpointDir.isEmpty) { throw new Exception("Checkpoint directory has not been set in the SparkContext") } else if (checkpointData.isEmpty) { checkpointData = Some(new RDDCheckpointData(this)) // 创建RDDCheckpointData, 记录关于CP的所有信息 checkpointData.get.markForCheckpoint() // 标记为Marked } } /** * Performs the checkpointing of this RDD by saving this. It is called by the DAGScheduler * after a job using this RDD has completed (therefore the RDD has been materialized and * potentially stored in memory). doCheckpoint() is called recursively on the parent RDDs. */ private[spark] def doCheckpoint() { if (!doCheckpointCalled) { doCheckpointCalled = true if (checkpointData.isDefined) { checkpointData.get.doCheckpoint() // 调用RDDCheckpointData.doCheckpoint } else { dependencies.foreach(_.rdd.doCheckpoint()) // 当checkpointData没有被创建, 就checkpoint所有的父RDD } } } /** * Changes the dependencies of this RDD from its original parents to a new RDD (`newRDD`) * created from the checkpoint file, and forget its old dependencies and partitions. */ private[spark] def markCheckpointed(checkpointRDD: RDD[_]) { clearDependencies() // dependencies_ = null partitions_ = null deps = null // Forget the constructor argument for dependencies too } /** * Compute an RDD partition or read it from a checkpoint if the RDD is checkpointing. */ private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext): Iterator[T] = { if (isCheckpointed) { firstParent[T].iterator(split, context) } else { compute(split, context) } } RDDCheckpointData /** * Enumeration to manage state transitions of an RDD through checkpointing * [ Initialized --> marked for checkpointing --> checkpointing in progress --> checkpointed ] */ private[spark] object CheckpointState extends Enumeration { // 定义Checkpoint过程中的各个状态 type CheckpointState = Value val Initialized, MarkedForCheckpoint, CheckpointingInProgress, Checkpointed = Value } /** * This class contains all the information related to RDD checkpointing. Each instance of this class * is associated with a RDD. It manages process of checkpointing of the associated RDD, as well as, * manages the post-checkpoint state by providing the updated partitions, iterator and preferred locations * of the checkpointed RDD. */ private[spark] class RDDCheckpointData[T: ClassManifest](rdd: RDD[T]) extends Logging with Serializable { import CheckpointState._ // The checkpoint state of the associated RDD. var cpState = Initialized // cp状态,先设为Initialized // The file to which the associated RDD has been checkpointed to @transient var cpFile: Option[String] = None // The CheckpointRDD created from the checkpoint file, that is, the new parent the associated RDD. var cpRDD: Option[RDD[T]] = None // Mark the RDD for checkpointing def markForCheckpoint() { RDDCheckpointData.synchronized { // 先mark, 防止两个同时做cp if (cpState == Initialized) cpState = MarkedForCheckpoint } } // Is the RDD already checkpointed def isCheckpointed: Boolean = { RDDCheckpointData.synchronized { cpState == Checkpointed } } // Get the file to which this RDD was checkpointed to as an Option def getCheckpointFile: Option[String] = { RDDCheckpointData.synchronized { cpFile } } // Do the checkpointing of the RDD. Called after the first job using that RDD is over. def doCheckpoint() { // If it is marked for checkpointing AND checkpointing is not already in progress, // then set it to be in progress, else return RDDCheckpointData.synchronized { if (cpState == MarkedForCheckpoint) { // 只有已经是MarkedForCheckpoint状态, 才能继续CP cpState = CheckpointingInProgress } else { return } } // Create the output path for the checkpoint val path = new Path(rdd.context.checkpointDir.get, "rdd-" + rdd.id) val fs = path.getFileSystem(new Configuration()) if (!fs.mkdirs(path)) { throw new SparkException("Failed to create checkpoint path " + path) } // Save to file, and reload it as an RDD rdd.context.runJob(rdd, CheckpointRDD.writeToFile(path.toString) _) // 将RDD存入文件, 完成cp val newRDD = new CheckpointRDD[T](rdd.context, path.toString) // 将RDD从文件重新载入 // Change the dependencies and partitions of the RDD RDDCheckpointData.synchronized { cpFile = Some(path.toString) cpRDD = Some(newRDD) rdd.markCheckpointed(newRDD) // 调用rdd.markCheckpointed清除deps cpState = Checkpointed // 将状态置为Checkpointed RDDCheckpointData.clearTaskCaches() logInfo("Done checkpointing RDD " + rdd.id + ", new parent is RDD " + newRDD.id) } } // Get preferred location of a split after checkpointing def getPreferredLocations(split: Partition): Seq[String] = { RDDCheckpointData.synchronized { cpRDD.get.preferredLocations(split) } } def getPartitions: Array[Partition] = { RDDCheckpointData.synchronized { cpRDD.get.partitions } } def checkpointRDD: Option[RDD[T]] = { RDDCheckpointData.synchronized { cpRDD } } } CheckpointRDD private[spark] class CheckpointRDDPartition(val index: Int) extends Partition {} /** * This RDD represents a RDD checkpoint file (similar to HadoopRDD). */ private[spark] class CheckpointRDD[T: ClassManifest](sc: SparkContext, val checkpointPath: String) extends RDD[T](sc, Nil) { @transient val fs = new Path(checkpointPath).getFileSystem(sc.hadoopConfiguration) override def getPartitions: Array[Partition] = { // 应该是每个partitions是一个文件, 所以就是看cp目录里面的文件个数 val cpath = new Path(checkpointPath) val numPartitions = // listStatus can throw exception if path does not exist. if (fs.exists(cpath)) { val dirContents = fs.listStatus(cpath) val partitionFiles = dirContents.map(_.getPath.toString).filter(_.contains("part-")).sorted val numPart = partitionFiles.size if (numPart > 0 && (! partitionFiles(0).endsWith(CheckpointRDD.splitIdToFile(0)) || ! partitionFiles(numPart-1).endsWith(CheckpointRDD.splitIdToFile(numPart-1)))) { throw new SparkException("Invalid checkpoint directory: " + checkpointPath) } numPart } else 0 Array.tabulate(numPartitions)(i => new CheckpointRDDPartition(i)) } checkpointData = Some(new RDDCheckpointData[T](this)) checkpointData.get.cpFile = Some(checkpointPath) override def getPreferredLocations(split: Partition): Seq[String] = { val status = fs.getFileStatus(new Path(checkpointPath, CheckpointRDD.splitIdToFile(split.index))) val locations = fs.getFileBlockLocations(status, 0, status.getLen) locations.headOption.toList.flatMap(_.getHosts).filter(_ != "localhost") } override def compute(split: Partition, context: TaskContext): Iterator[T] = { val file = new Path(checkpointPath, CheckpointRDD.splitIdToFile(split.index)) CheckpointRDD.readFromFile(file, context) // compute就是从cp文件里面把RDD读取出来 } override def checkpoint() { // Do nothing. CheckpointRDD should not be checkpointed. } } 本文章摘自博客园，原文发布日期：2014-01-10
2017-05-01
优秀的个人博客,低调大师

Spark源码分析 – SparkEnv

SparkEnv在两个地方会被创建, 由于SparkEnv中包含了很多重要的模块, 比如BlockManager, 所以SparkEnv很重要 Driver端, 在SparkContext初始化的时候, SparkEnv会被创建 // Create the Spark execution environment (cache, map output tracker, etc) private[spark] val env = SparkEnv.createFromSystemProperties( "<driver>", // 表示是driver, 下面的executor则是executorid System.getProperty("spark.driver.host"), System.getProperty("spark.driver.port").toInt, true, isLocal) SparkEnv.set(env) Executor端, 在executor初始化时被创建 // Initialize Spark environment (using system properties read above) val env = SparkEnv.createFromSystemProperties(executorId, slaveHostname, 0, false, false) SparkEnv.set(env) SparkEnv Class 用于hold所有Spark运行时的环境对象, serializer, Akka actor system, block manager, and map output tracker等 /** * Holds all the runtime environment objects for a running Spark instance (either master or worker), * including the serializer, Akka actor system, block manager, map output tracker, etc. Currently * Spark code finds the SparkEnv through a thread-local variable, so each thread that accesses these * objects needs to have the right SparkEnv set. You can get the current environment with * SparkEnv.get (e.g. after creating a SparkContext) and set it with SparkEnv.set. */ class SparkEnv ( val executorId: String, val actorSystem: ActorSystem, val serializerManager: SerializerManager, val serializer: Serializer, val closureSerializer: Serializer, val cacheManager: CacheManager, val mapOutputTracker: MapOutputTracker, val shuffleFetcher: ShuffleFetcher, val broadcastManager: BroadcastManager, val blockManager: BlockManager, val connectionManager: ConnectionManager, val httpFileServer: HttpFileServer, val sparkFilesDir: String, val metricsSystem: MetricsSystem) { } SparkEnv Object scala使用伴生object当作类接口 除了基本的get和set 就是在createFromSystemProperties中创建了一堆很关键的对象 object SparkEnv extends Logging { private val env = new ThreadLocal[SparkEnv] // ThreadLocal,所以每个线程各访问各的 @volatile private var lastSetSparkEnv : SparkEnv = _ // 缓存最新更新的SparkEnv,并且volatile,便于其他线程获得 def set(e: SparkEnv) { lastSetSparkEnv = e env.set(e) } /** * Returns the ThreadLocal SparkEnv, if non-null. Else returns the SparkEnv * previously set in any thread. */ def get: SparkEnv = { Option(env.get()).getOrElse(lastSetSparkEnv) // 没有local时, 可以用lastSetSparkEnv } /** * Returns the ThreadLocal SparkEnv. */ def getThreadLocal : SparkEnv = { env.get() // 只取到local的 } def createFromSystemProperties( executorId: String, hostname: String, port: Int, isDriver: Boolean, isLocal: Boolean): SparkEnv = { val (actorSystem, boundPort) = AkkaUtils.createActorSystem("spark", hostname, port) val classLoader = Thread.currentThread.getContextClassLoader // Create an instance of the class named by the given Java system property, or by // defaultClassName if the property is not set, and return it as a T def instantiateClass[T](propertyName: String, defaultClassName: String): T = { val name = System.getProperty(propertyName, defaultClassName) Class.forName(name, true, classLoader).newInstance().asInstanceOf[T] } val serializerManager = new SerializerManager val serializer = serializerManager.setDefault( System.getProperty("spark.serializer", "org.apache.spark.serializer.JavaSerializer")) val closureSerializer = serializerManager.get( System.getProperty("spark.closure.serializer", "org.apache.spark.serializer.JavaSerializer")) val connectionManager = blockManager.connectionManager val broadcastManager = new BroadcastManager(isDriver) val cacheManager = new CacheManager(blockManager) // BlockManager val blockManagerMaster = new BlockManagerMaster(registerOrLookup( // registerOrLookup表示只有在master上创建Actor对象, slave上只是创建ref "BlockManagerMaster", new BlockManagerMasterActor(isLocal))) val blockManager = new BlockManager(executorId, actorSystem, blockManagerMaster, serializer) // MapOutputTracker val mapOutputTracker = new MapOutputTracker() mapOutputTracker.trackerActor = registerOrLookup( // 同样只有在master创建actor对象 "MapOutputTracker", new MapOutputTrackerActor(mapOutputTracker)) // ShuffleFetcher val shuffleFetcher = instantiateClass[ShuffleFetcher]( "spark.shuffle.fetcher", "org.apache.spark.BlockStoreShuffleFetcher") val httpFileServer = new HttpFileServer() httpFileServer.initialize() System.setProperty("spark.fileserver.uri", httpFileServer.serverUri) val metricsSystem = if (isDriver) { MetricsSystem.createMetricsSystem("driver") } else { MetricsSystem.createMetricsSystem("executor") } metricsSystem.start() new SparkEnv( executorId, actorSystem, serializerManager, serializer, closureSerializer, cacheManager, mapOutputTracker, shuffleFetcher, broadcastManager, blockManager, connectionManager, httpFileServer, sparkFilesDir, metricsSystem) } } 本文章摘自博客园，原文发布日期：2014-01-13
2017-05-01
优秀的个人博客,低调大师

Android AsyncTask 源码解析

AsyncTask中的几个方法才能完成对任务的定制。经常需要去重写的方法有以下四个： 1.onPreExecute() 这个方法会在后台任务开始执行之间调用，用于进行一些界面上的初始化操作，比如显示一个进度条对话框等。 2.doInBackground(Params...) 这个方法中的所有代码都会在子线程中运行，我们应该在这里去处理所有的耗时任务。任务一旦完成就可以通过return语句来将任务的执行结果进行返回，如果AsyncTask的第三个泛型参数指定的是Void，就可以不返回任务执行结果。注意，在这个方法中是不可以进行UI操作的，如果需要更新UI元素，比如说反馈当前任务的执行进度，可以调用publishProgress(Progress...)方法来完成。 3.onProgressUpdate(Progress...) 当在后台任务中调
2017-01-15

资源下载

更多资源
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2026-09-24 大小: 189MB 下载: 9次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-08-21 大小: 1.42GB 下载: 6次
Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

时间: 2026-08-21 大小: 9.84MB 下载: 40次
WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源，继承了IntelliJ IDEA强大的JS部分的功能。

时间: 2025-12-13 大小: 249.54MB 下载: 22次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册