Spark Core源码精读计划3 | SparkContext辅助属性及后初始化

本文主要是介绍Spark Core源码精读计划3 | SparkContext辅助属性及后初始化,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

推荐阅读

《关于MQ面试的几件小事 | 消息队列的用途、优缺点、技术选型》         《关于MQ面试的几件小事 | 如何保证消息队列高可用和幂等》 《关于MQ面试的几件小事 | 如何保证消息不丢失》 《关于MQ面试的几件小事 | 如何保证消息按顺序执行》 《关于MQ面试的几件小事 | 消息积压在消息队列里怎么办》 《关于Redis的几件小事 | 使用目的与问题及线程模型》 《关于Redis的几件小事 | Redis的数据类型/过期策略/内存淘汰》 《关于Redis的几件小事 | 高并发和高可用》 《关于Redis的几件小事 | 持久化/缓存雪崩与穿透》 《关于Redis的几件小事 | 缓存与数据库双写时的数据一致性》 《关于Redis的几件小事 | 并发竞争和Cluster模式》 本文适用于知识共享-署名-相同方式共享(CC-BY-SA)3.0协议 目录
  • 前言

  • SparkContext中的辅助属性

    • creationSite

    • allowMultipleContexts

    • startTime & stopped

    • addedFiles/addedJars & _files/_jars

    • persistentRdds

    • executorEnvs & _executorMemory & _sparkUser

    • checkpointDir

    • localProperties

    • _eventLogDir & _eventLogCodec

    • _applicationId & _applicationAttemptId

    • _shutdownHookRef

    • nextShuffleId & nextRddId

  • SparkContext后初始化

    • setupAndStartListenerBus()方法

    • postEnvironmentUpdate()方法

    • postApplicationStart()方法

    • 其他事项

  • 总结


前言

在文章#2中,我们了解了SparkContext的主体部分,即组件初始化。除了它之外,SparkContext中还有一些与其内部机制紧密相关的属性,下文为了简单,就将它们称为“辅助属性”。另外,在组件初始化完成后,还有一些善后工作,即后初始化(Post-init)。本文就来研究这两块内容。

SparkContext中的辅助属性

仿照文章#2中的方式,仍然先将我们要关注的这些属性整理出来。 代码#3.1 - SparkContext中的辅助属性
 private val creationSite: CallSite = Utils.getCallSite()  private val allowMultipleContexts: Boolean = config.getBoolean("spark.driver.allowMultipleContexts", false)  val startTime = System.currentTimeMillis()  private[spark] val stopped: AtomicBoolean = new AtomicBoolean(false)  private[spark] val addedFiles = new ConcurrentHashMap[String, Long]().asScala  private[spark] val addedJars = new ConcurrentHashMap[String, Long]().asScala  private[spark] val persistentRdds = {    val map: ConcurrentMap[Int, RDD[_]] = new MapMaker().weakValues().makeMap[Int, RDD[_]]()    map.asScala  }  private[spark] val executorEnvs = HashMap[String, String]()  val sparkUser = Utils.getCurrentUserName()  private[spark] var checkpointDir: Option[String] = None  protected[spark] val localProperties = new InheritableThreadLocal[Properties] {    override protected def childValue(parent: Properties): Properties = {      SerializationUtils.clone(parent)    }    override protected def initialValue(): Properties = new Properties()  }  private val nextShuffleId = new AtomicInteger(0)  private val nextRddId = new AtomicInteger(0)
private var _eventLogDir: Option[URI] = None private var _eventLogCodec: Option[String] = None private var _executorMemory: Int = _ private var _applicationId: String = _ private var _applicationAttemptId: Option[String] = None private var _jars: Seq[String] = _ private var _files: Seq[String] = _ private var _shutdownHookRef: AnyRef = _


以下划线开头的字段如同代码#2.2中一样,也有对应的Getter方法。 为了节省篇幅,就不列出来了。 下面按照它们初始化的顺序和相关性来介绍,必要时仍然会附上一些源码。


creationSite
creationSite指示SparkContext是在哪里创建的。 CallSite是个简单的数据结构,只有shortForm与longForm两个属性,用来描述代码的位置。 Utils.getCallSite()方法遍历当前线程的线程栈,并找到最后一个(即最靠近栈顶的)Spark方法调用,与最先一个(即最靠近栈底的)用户方法调用,将它们的短形式和长形式包装在CallSite中返回。 有兴趣的看官可以自行去看这个方法的源代码,不难。 以代码#0.1的WordCount为例,运行时打上断点,观察creationSite的内容如下图。


640?wx_fmt=other


allowMultipleContexts
allowMultipleContexts指示是否允许一个JVM(即一个Application)内存在多个活动的SparkContext实例。 它由spark.driver.allowMultipleContexts参数控制,默认为false,即只允许存在一个活动的SparkContext实例,如果有多个就会抛出异常。 设为true的话,在有多个活动的SparkContext时只会输出警告。 关于它在下一篇文章中还会涉及到,这里就不多说了。


startTime & stopped
startTime指示SparkContext启动时的时间戳。 stopped则指示SparkContext是否停止,它采用AtomicBoolean类型。


addedFiles/addedJars & _files/_jars
Spark支持在提交应用时,附带用户自定义的其他文件与JAR包。 addedFiles和addedJars是两个ConcurrentHashMap,用来维护自定义文件及JAR包的URL路径,及它们被加入ConcurrentHashMap当时的时间戳。 _files与_jars则接受Spark配置中定义的文件或JAR包路径。 由于它们的逻辑基本相同, 下面以JAR包为例来看一下代码。 代码#3.2 - 构造方法中自定义JAR包的初始化
    _jars = Utils.getUserJars(_conf)    if (jars != null) {      jars.foreach(addJar)    }

首先用Utils.getUserJars()方法从SparkConf的spark.jars配置项中取出路径组成的序列,然后分别调用addJar()方法。 代码#3.3 - o.a.s.SparkContext.addJar()方法
  def addJar(path: String) {    def addJarFile(file: File): String = {      try {        if (!file.exists()) {          throw new FileNotFoundException(s"Jar ${file.getAbsolutePath} not found")        }        if (file.isDirectory) {          throw new IllegalArgumentException(            s"Directory ${file.getAbsoluteFile} is not allowed for addJar")        }        env.rpcEnv.fileServer.addJar(file)      } catch {        case NonFatal(e) =>          logError(s"Failed to add $path to Spark environment", e)          null      }    }
if (path == null) { logWarning("null specified as parameter to addJar") } else { val key = if (path.contains("\\")) { addJarFile(new File(path)) } else { val uri = new URI(path) Utils.validateURL(uri) uri.getScheme match { case null => addJarFile(new File(uri.getRawPath)) case "file" => addJarFile(new File(uri.getPath)) case "local" => "file:" + uri.getPath case _ => path } } if (key != null) { val timestamp = System.currentTimeMillis if (addedJars.putIfAbsent(key, timestamp).isEmpty) { logInfo(s"Added JAR $path at $key with timestamp $timestamp") postEnvironmentUpdate() } } } }


addJar()方法检查JAR包路径的合法性和类型,然后调用RpcEnv中的RpcEnvFileServer.addJar()方法,将JAR包加进RPC环境中。 在该方法的最后还调用了postEnvironmentUpdate(),用来更新执行环境,这属于后初始化逻辑的一部分,下一节会讲到。


persistentRdds
Spark支持RDD的持久化,可以持久化到内存或磁盘。 persistentRdds维护的是持久化RDD的ID与其弱引用的映射关系。 通过RDD内自带的cache()/persist()/unpersist()方法可以持久化与反持久化一个RDD,它们最终调用的是SparkContext.persistRDD()/unpersistRDD()内部方法。 代码#3.4 - o.a.s.SparkContext.persistRDD()与unpersistRDD()方法
  private[spark] def persistRDD(rdd: RDD[_]) {    persistentRdds(rdd.id) = rdd  }
private[spark] def unpersistRDD(rddId: Int, blocking: Boolean = true) { env.blockManager.master.removeRdd(rddId, blocking) persistentRdds.remove(rddId) listenerBus.post(SparkListenerUnpersistRDD(rddId)) }

executorEnvs & _executorMemory & _sparkUser
executorEnvs是一个HashMap,用来存储需要传递给Executor的环境变量。 _executorMemory与_sparkUser就是其中之二,分别代表Executor内存大小和当前启动SparkContext的用户名。


代码#3.5 - 构造方法中Executor环境变量的初始化
    _executorMemory = _conf.getOption("spark.executor.memory")      .orElse(Option(System.getenv("SPARK_EXECUTOR_MEMORY")))      .orElse(Option(System.getenv("SPARK_MEM"))      .map(warnSparkMem))      .map(Utils.memoryStringToMb)      .getOrElse(1024)
for { (envKey, propKey) <- Seq(("SPARK_TESTING", "spark.testing")) value <- Option(System.getenv(envKey)).orElse(Option(System.getProperty(propKey)))} { executorEnvs(envKey) = value } Option(System.getenv("SPARK_PREPEND_CLASSES")).foreach { v => executorEnvs("SPARK_PREPEND_CLASSES") = v } executorEnvs("SPARK_EXECUTOR_MEMORY") = executorMemory + "m" executorEnvs ++= _conf.getExecutorEnv executorEnvs("SPARK_USER") = sparkUser


可见,Executor内存可以通过spark.executor.memory配置项、SPARK_EXECUTOR_MEMORY环境变量、SPARK_MEM环境变量指定,优先级依次降低,且默认大小是1GB。 用户名是通过Utils.getCurrentUserName()方法获得的。


checkpointDir
checkpointDir指定集群状态下,RDD检查点在HDFS上保存的目录。 检查点的存在是为了当计算过程出错时,能够快速恢复,而不必从头重新计算。 SparkContext提供了setCheckpointDir()方法用来设定检查点目录,如下。 代码#3.6 - o.a.s.SparkContext.setCheckpointDir()方法
 
 def setCheckpointDir(directory: String) {    if (!isLocal && Utils.nonLocalPaths(directory).isEmpty) {      logWarning("Spark is not running in local mode, therefore the checkpoint directory " +        s"must not be on the local filesystem. Directory '$directory' " +        "appears to be on the local filesystem.")    }
checkpointDir = Option(directory).map { dir => val path = new Path(dir, UUID.randomUUID().toString) val fs = path.getFileSystem(hadoopConfiguration) fs.mkdirs(path) fs.getFileStatus(path).getPath.toString } }

localProperties
localProperties用于维护一个Properties数据类型的线程本地变量。 它是InheritableThreadLocal类型,继承自ThreadLocal,在后者的基础上允许本地变量从父线程到子线程的继承,也就是该Properties会沿着线程栈传递下去。


_eventLogDir & _eventLogCodec
这两个属性与EventLoggingListener相关。 EventLoggingListener打开时,事件日志会写入_eventLogDir指定的目录,可以用spark.eventLog.dir参数设置。 _eventLogCodec指定事件日志的压缩算法,当通过spark.eventLog.compress参数启用压缩后,就根据spark.io.compression.codec参数配置压缩算法,目前支持lz4、lzf、snappy、zstd四种。


_applicationId & _applicationAttemptId
这两个ID都是TaskScheduler初始化完毕并启动之后才分配的。 TaskScheduler启动之后,应用代码的逻辑才真正被执行,并且可能会进行多次尝试。 在SparkUI、BlockManager和EventLoggingListener初始化时,也会用到它们。 代码#3.7 - 构造方法中_applicationId与_applicationAttemptId的初始化
 
   _applicationId = _taskScheduler.applicationId()    _applicationAttemptId = taskScheduler.applicationAttemptId()
_shutdownHookRef
它用来定义SparkContext的关闭钩子,主要是在JVM退出时,显式地执行SparkContext.stop()方法,以防止用户忘记而留下烂摊子。 这实际上是后初始化逻辑,在下面的代码#3.8中会出现。


nextShuffleId & nextRddId
这两个ID都是AtomicInteger类型。 Shuffle和RDD都需要唯一ID来进行标识,并且它们是递增的。 在代码#3.4中已经出现过了RDD ID。

SparkContext后初始化

在文章#2的ContextCleaner初始化之后,还有一小部分后初始化逻辑,其代码如下所示。 代码#3.8 - SparkContext后初始化逻辑
    setupAndStartListenerBus()    postEnvironmentUpdate()    postApplicationStart()
_taskScheduler.postStartHook() _env.metricsSystem.registerSource(_dagScheduler.metricsSource) _env.metricsSystem.registerSource(new BlockManagerSource(_env.blockManager)) _executorAllocationManager.foreach { e => _env.metricsSystem.registerSource(e.executorAllocationManagerSource) }
logDebug("Adding shutdown hook") // force eager creation of logger _shutdownHookRef = ShutdownHookManager.addShutdownHook( ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY) { () => logInfo("Invoking stop() from shutdown hook") stop() }
    SparkContext.setActiveContext(this, allowMultipleContexts)
它的主要逻辑在开头的三个方法中,下面来逐一看它们的代码。


setupAndStartListenerBus()方法
代码#3.9 - o.a.s.SparkContext.setupAndStartListenerBus()方法
  private def setupAndStartListenerBus(): Unit = {    try {      conf.get(EXTRA_LISTENERS).foreach { classNames =>        val listeners = Utils.loadExtensions(classOf[SparkListenerInterface], classNames, conf)        listeners.foreach { listener =>          listenerBus.addToSharedQueue(listener)          logInfo(s"Registered listener ${listener.getClass().getName()}")        }      }    } catch {      case e: Exception =>        try {          stop()        } finally {          throw new SparkException(s"Exception when registering SparkListener", e)        }    }
listenerBus.start(this, _env.metricsSystem) _listenerBusStarted = true }

这个方法用于注册自定义的监听器,并最终启动LiveListenerBus。 自定义监听器都实现了SparkListener特征,通过spark.extraListeners配置参数来指定。 然后调用Utils.loadExtensions()方法,通过反射来构建自定义监听器的实例,并将它们注册到LiveListenerBus。


postEnvironmentUpdate()方法
代码#3.10 - o.a.s.SparkContext.postEnvironmentUpdate()方法
  private def postEnvironmentUpdate() {    if (taskScheduler != null) {      val schedulingMode = getSchedulingMode.toString      val addedJarPaths = addedJars.keys.toSeq      val addedFilePaths = addedFiles.keys.toSeq      val environmentDetails = SparkEnv.environmentDetails(conf, schedulingMode, addedJarPaths,        addedFilePaths)      val environmentUpdate = SparkListenerEnvironmentUpdate(environmentDetails)      listenerBus.post(environmentUpdate)    }  }

该方法在添加自定义文件和JAR包时也都有调用,因为添加的资源会对程序的执行环境造成影响。 它会取得当前的自定义文件和JAR包列表,以及Spark配置、调度方式,然后通过SparkEnv.environmentDetails()方法再取得JVM参数、Java系统属性等,一同封装成SparkListenerEnvironmentUpdate事件,并投递给事件总线。


postApplicationStart()方法
代码#3.11 - o.a.s.SparkContext.postApplicationStart()方法
  private def postApplicationStart() {    listenerBus.post(SparkListenerApplicationStart(appName, Some(applicationId),      startTime, sparkUser, applicationAttemptId, schedulerBackend.getDriverLogUrls))  }

这个方法比较简单,就是向事件总线投递SparkListenerApplicationStart事件,表示Application已经启动。


其他事项
在这三个方法之后的其他事项如下。
  • 调用TaskScheduler.postStartHook()方法,等待SchedulerBackend初始化完毕。

  • 在度量系统中注册DAGScheduler、BlockManager、ExecutionAllocationManager的度量源,以收集它们的监控数据。

  • 添加关闭钩子,这个在之前已经提过了,不再赘述。

  • 调用伴生对象中的setActiveContext()方法,将当前SparkContext设为活动的。


总结

本文通过梳理SparkContext中的多个辅助属性,进一步了解了一些细节特性,如外部文件和JAR包的初始化、RDD持久化和检查点等。 在SparkContext构造方法的最后,还会执行一些扫尾的工作,如启动事件总线、更新执行环境等。 SparkContext除了初始化之外,还对外提供了不少通用的功能,如生成RDD,产生广播变量与累加器,启动Job等等。 另外,SparkContext类也有伴生对象,里面维护了一些常用的逻辑。 下一篇文章作为SparkContext概况的收尾,就来研究这些剩下的东西。


— THE END —

640?wx_fmt=png

这篇关于Spark Core源码精读计划3 | SparkContext辅助属性及后初始化的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1143210

相关文章

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很

浅析CSS 中z - index属性的作用及在什么情况下会失效

《浅析CSS中z-index属性的作用及在什么情况下会失效》z-index属性用于控制元素的堆叠顺序,值越大,元素越显示在上层,它需要元素具有定位属性(如relative、absolute、fi... 目录1. z-index 属性的作用2. z-index 失效的情况2.1 元素没有定位属性2.2 元素处

Spring 中 BeanFactoryPostProcessor 的作用和示例源码分析

《Spring中BeanFactoryPostProcessor的作用和示例源码分析》Spring的BeanFactoryPostProcessor是容器初始化的扩展接口,允许在Bean实例化前... 目录一、概览1. 核心定位2. 核心功能详解3. 关键特性二、Spring 内置的 BeanFactory

Spring组件初始化扩展点BeanPostProcessor的作用详解

《Spring组件初始化扩展点BeanPostProcessor的作用详解》本文通过实战案例和常见应用场景详细介绍了BeanPostProcessor的使用,并强调了其在Spring扩展中的重要性,感... 目录一、概述二、BeanPostProcessor的作用三、核心方法解析1、postProcessB

HTML5 data-*自定义数据属性的示例代码

《HTML5data-*自定义数据属性的示例代码》HTML5的自定义数据属性(data-*)提供了一种标准化的方法在HTML元素上存储额外信息,可以通过JavaScript访问、修改和在CSS中使用... 目录引言基本概念使用自定义数据属性1. 在 html 中定义2. 通过 JavaScript 访问3.

CSS模拟 html 的 title 属性(鼠标悬浮显示提示文字效果)

《CSS模拟html的title属性(鼠标悬浮显示提示文字效果)》:本文主要介绍了如何使用CSS模拟HTML的title属性,通过鼠标悬浮显示提示文字效果,通过设置`.tipBox`和`.tipBox.tipContent`的样式,实现了提示内容的隐藏和显示,详细内容请阅读本文,希望能对你有所帮助... 效

解读为什么@Autowired在属性上被警告,在setter方法上不被警告问题

《解读为什么@Autowired在属性上被警告,在setter方法上不被警告问题》在Spring开发中,@Autowired注解常用于实现依赖注入,它可以应用于类的属性、构造器或setter方法上,然... 目录1. 为什么 @Autowired 在属性上被警告?1.1 隐式依赖注入1.2 IDE 的警告:

HTML5中下拉框<select>标签的属性和样式详解

《HTML5中下拉框<select>标签的属性和样式详解》在HTML5中,下拉框(select标签)作为表单的重要组成部分,为用户提供了一个从预定义选项中选择值的方式,本文将深入探讨select标签的... 在html5中,下拉框(<select>标签)作为表单的重要组成部分,为用户提供了一个从预定义选项中

C++初始化数组的几种常见方法(简单易懂)

《C++初始化数组的几种常见方法(简单易懂)》本文介绍了C++中数组的初始化方法,包括一维数组和二维数组的初始化,以及用new动态初始化数组,在C++11及以上版本中,还提供了使用std::array... 目录1、初始化一维数组1.1、使用列表初始化(推荐方式)1.2、初始化部分列表1.3、使用std::

在Java中使用ModelMapper简化Shapefile属性转JavaBean实战过程

《在Java中使用ModelMapper简化Shapefile属性转JavaBean实战过程》本文介绍了在Java中使用ModelMapper库简化Shapefile属性转JavaBean的过程,对比... 目录前言一、原始的处理办法1、使用Set方法来转换2、使用构造方法转换二、基于ModelMapper