首页 文章 精选 留言 我的

精选列表

搜索[认知分析]，共10000篇文章
优秀的个人博客,低调大师

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 源码分析 -- 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
优秀的个人博客,低调大师

Hadoop的RPC分析

一、基础知识 原理http://www.cnblogs.com/edisonchou/p/4285817.html，这个谢了一些rpc与hadoop的例子。 用到了java的动态代理，服务端实现一个接口，客户端得到这个接口的实现类，客户端通过自定义的versionID来标志一对服务端和客户端。 二、代码 2.1 接口 package rpc; import org.apache.hadoop.ipc.VersionedProtocol; //默认情况下，不同版本号的RPC Client和Server之间不能相互通信，因此客户端和服务端通过版本号标识。 public interface RPCable extends VersionedProtocol { public static final long versionID = 10010; public String sayHi(String name) ; } 我原来没继承VersionedProtocol接口，也成功了。接口不能在实现接口 2.2 服务端 package rpc; import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.ipc.ProtocolSignature; import org.apache.hadoop.ipc.RPC; import org.apache.hadoop.ipc.RPC.Server; //动态代理：客户端和服务端同时实现一个接口 //可以在linux和win下 //不同main进程下相互调用 //是不同进程见的方法调用 //底层走的是jav的rmi public class RPCServer implements RPCable{ public String sayHi(String name) { return "HI！"+name; } public static void main(String[] args) throws Exception { // TODO Auto-generated method stub Configuration conf = new Configuration(); //方法的调用连 因为每次调用返回的是this Server server = new RPC.Builder(conf).setProtocol(RPCable.class). setInstance(new RPCServer()).setBindAddress("localhost"). setPort(9527).build(); server.start(); } @Override public ProtocolSignature getProtocolSignature(String arg0, long arg1, int arg2) throws IOException { // TODO Auto-generated method stub return null; } @Override public long getProtocolVersion(String arg0, long arg1) throws IOException { // TODO Auto-generated method stub return 0; }; } 我原来没继承VersionedProtocol类，那么服务端不需要下面的两个方法，继承了以后必须写。 2.3 客户端 package rpc; import java.io.IOException; import java.net.InetSocketAddress; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.ipc.RPC; //打印在客户端 拼接字符串确实在服务端 public class RPCClient { public static void main(String[] args) throws Exception { // TODO Auto-generated method stub //协议一般是.class RPCable proxy = RPC.getProxy(RPCable.class, 10010, new InetSocketAddress("localhost",9527), new Configuration()); String res = proxy.sayHi("！！！！！！！！！！！！"); System.out.println(res); RPC.stopProxy(proxy); } } 三、结束 rpc不受系统、地理位置的限制。 不相信的话可以在和护短打断点，一步一步发现进入了；额服务器端。 做的时候如果不知道使用那个类哭下的东西，可以打开对应类的class文件，可能是黑色，不是源码，那么点击attach source找到hadoop的源码包就可以了，导进去就看到的是普通的和咱的java类似的文件，可以看看里面的蚕食来自那个包。
2016-12-05

资源下载

更多资源
Nacos

Nacos

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

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

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2026-08-21 大小: 5MB 下载: 2次
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次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

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

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

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册