首页 文章 精选 留言 我的

精选列表

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

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

Kafka性能测试分析

首先要特别感谢赵崇贺同学利用业余时间进行的压测，才能为本文提供专业的测试数据 一、测试环境准备 �Cpu �内存 硬盘 Intel(R) Xeon(R) CPU E5520 @ 2.27GHz 32G 6T Kafka集群，服务器个数：3台 采用CMS垃圾回收 JVM运行参数 -Xmx1G -Xms1G -server -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true -Xloggc:/usr/local/kafka_2.10-0.8.2.2/bin/../logs/kafkaServer-gc.log -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dkafka.logs.dir=/usr/local/kafka_2.10-0.8.2.2/bin/../logs -Dlog4j.configuration=file:/usr/local/kafka_2.10-0.8.2.2/bin/../config/log4j.properties Kafka server端配置 broker.id=165 port=9092 host.name=hadoop165.kuaiyong.in num.network.threads=3 num.io.threads=8 socket.send.buffer.bytes=102400 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 log.dirs=/download/kafka-logs num.partitions=1 num.recovery.threads.per.data.dir=1 log.retention.hours=168 log.segment.bytes=1073741824 log.retention.check.interval.ms=300000 log.cleaner.enable=false zookeeper.connect=hadoop165.xxx.in:2181,hadoop166.xxx.in:2181,hadoop167.xxx.in:2181 zookeeper.connection.timeout.ms=6000 测试命令行如下 生产者： bin/kafka-producer-perf-test.sh --broker-list=hadoop02:9092 --messages 100000 --topic s1 --threads 10 --message-size 1000 --batch-size 200 --compression-codec 1 消费者 bin/kafka-consumer-perf-test.sh --zookeeper hadoop03:2181 --messages 500000 --topic s1 --threads 1 二、正常请求测试 1、Producer： 数据量：230万条记录 每次打包发送1000条数据 数据格式：采用压缩格式 测试结果 最大处理量：39.2501MB/S TPS：41156.6817条 2、Consumer 耗时：18秒 总体文件大小：2193.45MB 最大处理量：163.6659MB/S TPS：171616.1767条 三、压力请求测试 1、Producer 数据量：1000万条数据 每次打包发送1000条数据 数据格式：采用压缩格式 测试结果 耗时：242秒 总体文件大小：9536.74MB 最大处理量：39.2531MB/S TPS：41159.8856条 2、Consumer 耗时：70秒 总体文件大小：9536.74MB 最大处理量：145.4193MB/S TPS：152483.1887条 结论：1000万条下的压力测试，性能有所降低，估计瓶颈期在500万条左右
2016-03-01

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2026-04-11 大小: 211.28KB 下载: 56次
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

时间: 2026-08-21 大小: 1MB 下载: 10次
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次
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

用户登录
用户注册