本文针对于「Spark2.4」系列分析
代码语言：javascript复制/**
* Main entry point for Spark functionality. A SparkContext represents the connection to a Spark
* cluster, and can be used to create RDDs, accumulators and broadcast variables on that cluster.
* Spark功能的主要入口点。SparkContext表示与Spark集群的连接，可用于在该集群上创建RDD，累加器和广播变量。
*
* Only one SparkContext may be active per JVM. You must `stop()` the active SparkContext before
* creating a new one. This limitation may eventually be removed; see SPARK-2243 for more details.
* 每个JVM只能激活一个SparkContext。在创建新的SparkContext之前，您必须“停止（）”活动的SparkContext。
* 此限制可能最终会消除；有关更多详细信息，请参见SPARK-2243
*
* @param config a Spark Config object describing the application configuration. Any settings in
* this config overrides the default configs as well as system properties.
* 传入参数config是一个描述应用程序配置的Spark Config对象。此配置中的任何设置都会覆盖默认配置以及系统属性
*/
SparkDriver核心组件
须知
一、 CallSite创建
什么叫CallSite？CallSite有什么用？
代码语言：javascript复制/** CallSite represents a place in user code. It can have a short and a long form. */
CallSite表示用户代码中的一个位置。它可以有短的和长的形式。(最短栈、最长栈)
private[spark] case class CallSite(shortForm: String, longForm: String)
源码中通过「getCallSite()」 方法配置返回CallSite
参数示意：
参数英文名 | 参数含义 |
|---|---|
lastSparkMethod | 方法存入 |
firstUserFile | 类名存入 |
firstUserLine | 行号存入 |
源码如下：
代码语言：javascript复制 def getCallSite(skipClass: String => Boolean = sparkInternalExclusionFunction): CallSite = {
// Keep crawling up the stack trace until we find the first function not inside of the spark
// package. We track the last (shallowest) contiguous Spark method. This might be an RDD
// transformation, a SparkContext function (such as parallelize), or anything else that leads
// to instantiation of an RDD. We also track the first (deepest) user method, file, and line.
var lastSparkMethod = "<unknown>"
var firstUserFile = "<unknown>"
var firstUserLine = 0
var insideSpark = true
val callStack = new ArrayBuffer[String]() : "<unknown>"
Thread.currentThread.getStackTrace().foreach { ste: StackTraceElement =>
// When running under some profilers, the current stack trace might contain some bogus
// frames. This is intended to ensure that we don't crash in these situations by
// ignoring any frames that we can't examine.
if (ste != null && ste.getMethodName != null
&& !ste.getMethodName.contains("getStackTrace")) {
if (insideSpark) {
if (skipClass(ste.getClassName)) {
lastSparkMethod = if (ste.getMethodName == "<init>") {
// Spark method is a constructor; get its class name
ste.getClassName.substring(ste.getClassName.lastIndexOf('.') 1)
} else {
ste.getMethodName
}
callStack(0) = ste.toString // Put last Spark method on top of the stack trace.
} else {
if (ste.getFileName != null) {
firstUserFile = ste.getFileName
if (ste.getLineNumber >= 0) {
firstUserLine = ste.getLineNumber
}
}
callStack = ste.toString
insideSpark = false
}
} else {
callStack = ste.toString
}
}
}
val callStackDepth = System.getProperty("spark.callstack.depth", "20").toInt
val shortForm =
if (firstUserFile == "HiveSessionImpl.java") {
// To be more user friendly, show a nicer string for queries submitted from the JDBC
// server.
"Spark JDBC Server Query"
} else {
s"$lastSparkMethod at $firstUserFile:$firstUserLine"
}
val longForm = callStack.take(callStackDepth).mkString("n")
CallSite(shortForm, longForm)
}
客户端结果：
代码语言：javascript复制举例：WordCount例子中，获得数据如下
最短栈：SparkContext at MyWorkCount.scala:7
最长栈：org.apache.spark.SparkContext.<init>(SparkContext.scala:76)
com.spark.MyWorkCount$.main(MyWorkCount.scala:7)
com.spark.MyWorkCount.main(MyWorkCount.scala)
二、ActiveContext取舍
代码语言：javascript复制// If true, log warnings instead of throwing exceptions when multiple SparkContexts are active
// 如果为true，则在多个SparkContext处于活动状态时记录警告而不是引发异常 默认false
private val allowMultipleContexts: Boolean =
config.getBoolean("spark.driver.allowMultipleContexts", false)
// In order to prevent multiple SparkContexts from being active at the same time, mark this
// context as having started construction.
// NOTE: this must be placed at the beginning of the SparkContext constructor.
// 为了防止同时激活多个SparkContext，将此上下文标记为active。以防止多个SparkContext实例同时成为active级别的。
SparkContext.markPartiallyConstructed(this, allowMultipleContexts)
val startTime = System.currentTimeMillis()
private[spark] val stopped: AtomicBoolean = new AtomicBoolean(false)
//断言存活的SparkCntext
private[spark] def assertNotStopped(): Unit = {
if (stopped.get()) {
val activeContext = SparkContext.activeContext.get()
val activeCreationSite =
if (activeContext == null) {
"(No active SparkContext.)"
} else {
activeContext.creationSite.longForm
}
throw new IllegalStateException(
s"""Cannot call methods on a stopped SparkContext.
|This stopped SparkContext was created at:
|
|${creationSite.longForm}
|
|The currently active SparkContext was created at:
|
|$activeCreationSite
""".stripMargin)
}
}
正式篇
一、读取SparkConf、日志压缩配置
Spark配置类，配置已键值对形式存储，封装了一个ConcurrentHashMap类实例settings用于存储Spark的配置信息。
代码语言：javascript复制 //copy一份配置文件
_conf = config.clone()
//必要信息检查，验证提交配置项参数、提交方式
_conf.validateSettings()
//检查部署模式spark.master配置
if (!_conf.contains("spark.master")) {
throw new SparkException("A master URL must be set in your configuration")
}
//检查spark.app.name配置
if (!_conf.contains("spark.app.name")) {
throw new SparkException("An application name must be set in your configuration")
}
// log out spark.app.name in the Spark driver logs 打印应用程序名称
logInfo(s"Submitted application: $appName")
// System property spark.yarn.app.id must be set if user code ran by AM on a YARN cluster
// 如果用户代码由AM在YARN群集上运行，则必须设置系统属性spark.yarn.app.id
if (master == "yarn" && deployMode == "cluster" && !_conf.contains("spark.yarn.app.id")) {
throw new SparkException("Detected yarn cluster mode, but isn't running on a cluster. "
"Deployment to YARN is not supported directly by SparkContext. Please use spark-submit.")
}
// 检查日志配置
if (_conf.getBoolean("spark.logConf", false)) {
logInfo("Spark configuration:n" _conf.toDebugString)
}
// Set Spark driver host and port system properties. This explicitly sets the configuration
// instead of relying on the default value of the config constant.
// 设置Spark驱动程序主机和端口系统属性。这将显式设置配置而不依赖于config常量的默认值
_conf.set(DRIVER_HOST_ADDRESS, _conf.get(DRIVER_HOST_ADDRESS))
_conf.setIfMissing("spark.driver.port", "0")
_conf.set("spark.executor.id", SparkContext.DRIVER_IDENTIFIER)
//获取用户传入jar包
//在YARN模式下，它将返回一个空列表，因为YARN 具有自己的分发jar的机制。
_jars = Utils.getUserJars(_conf)
//获取用户传入的文件
_files = _conf.getOption("spark.files").map(_.split(",")).map(_.filter(_.nonEmpty))
.toSeq.flatten
//事件日志目录
_eventLogDir =
if (isEventLogEnabled) {
val unresolvedDir = conf.get("spark.eventLog.dir", EventLoggingListener.DEFAULT_LOG_DIR)
.stripSuffix("/")
Some(Utils.resolveURI(unresolvedDir))
} else {
None
}
//事件日志压缩 默认flase不压缩
_eventLogCodec = {
val compress = _conf.getBoolean("spark.eventLog.compress", false)
if (compress && isEventLogEnabled) {
Some(CompressionCodec.getCodecName(_conf)).map(CompressionCodec.getShortName)
} else {
None
}
}
二、初始化LiveListenerBus
SparkContext 中的事件总线，可以接收各种使用方的事件，并且异步传递Spark事件监听与SparkListeners监听器的注册。
代码语言：javascript复制 //创建生命周期监听总线
_listenerBus = new LiveListenerBus(_conf)
// Initialize the app status store and listener before SparkEnv is created so that it gets
// all events.
// 在创建SparkEnv之前 初始化 应用程序状态存储 和 侦听器，以便获取所有事件
_statusStore = AppStatusStore.createLiveStore(conf)
listenerBus.addToStatusQueue(_statusStore.listener.get)
三、创建SparkENV对象(DriverENV)
SparkContext中非常重要的类，它维护着Spark的执行环境，所有的线程都可以通过SparkContext访问到同一个SparkEnv对象。包含一些rpc创建……etc.
「LiveListenerBus」 生命周期监听总线
代码语言：javascript复制 // Create the Spark execution environment (cache, map output tracker, etc)
// 创建SparkEev 执行环境(cache, map输出追踪器, 等等)
_env = createSparkEnv(_conf, isLocal, listenerBus)
SparkEnv.set(_env)
// If running the REPL, register the repl's output dir with the file server.
//REPL-> “读取-求值-输出”循环(英语:Read-Eval-Print Loop,简称REPL)指的是一个简单的,交互式的编程环境
// 如果运行REPL，请向文件服务器注册repl的输出目录。
_conf.getOption("spark.repl.class.outputDir").foreach { path =>
val replUri = _env.rpcEnv.fileServer.addDirectory("/classes", new File(path))
_conf.set("spark.repl.class.uri", replUri)
}
--------------------------------------------------------------------------
private[spark] def createSparkEnv(
conf: SparkConf,
isLocal: Boolean,
listenerBus: LiveListenerBus): SparkEnv = {
SparkEnv.createDriverEnv(conf, isLocal, listenerBus, SparkContext.numDriverCores(master, conf))
}
四、初始化SparkStatusTracker
低级别的状态报告API，只能提供非常脆弱的一致性机制，对Job（作业）、Stage（阶段）的状态进行监控。
代码语言：javascript复制 //用于监视job和stage的进度
//注意SparkStatusTracker中API提供非常弱的一致性语义,在Active阶段中有可能返回'None'
_statusTracker = new SparkStatusTracker(this, _statusStore)
五、初始化ConsoleProgressBar
进度条 [stage1]====================>
代码语言：javascript复制 //说白了就是console print的那个线。。。。。。
_progressBar =
if (_conf.get(UI_SHOW_CONSOLE_PROGRESS) && !log.isInfoEnabled) {
Some(new ConsoleProgressBar(this))
} else {
None
}
--------------------------------------------------------------------
private def show(now: Long, stages: Seq[StageData]) {
val width = TerminalWidth / stages.size
val bar = stages.map { s =>
val total = s.numTasks
val header = s"[Stage ${s.stageId}:"
val tailer = s"(${s.numCompleteTasks} ${s.numActiveTasks}) / $total]"
val w = width - header.length - tailer.length
val bar = if (w > 0) {
val percent = w * s.numCompleteTasks / total
(0 until w).map { i =>
if (i < percent) "=" else if (i == percent) ">" else " "
}.mkString("")
} else {
""
}
header bar tailer
}.mkString("")
// only refresh if it's changed OR after 1 minute (or the ssh connection will be closed
// after idle some time)
if (bar != lastProgressBar || now - lastUpdateTime > 60 * 1000L) {
System.err.print(CR bar)
lastUpdateTime = now
}
lastProgressBar = bar
}
六、创建&初始化 Spark UI
Spark监控的web平台，提供了整个生命周期的监控包括任务、环境。
代码语言：javascript复制 //是否允许UI开启
_ui =
if (conf.getBoolean("spark.ui.enabled", true)) {
Some(SparkUI.create(Some(this), _statusStore, _conf, _env.securityManager, appName, "",
startTime))
} else {
// For tests, do not enable the UI
None
}
// Bind the UI before starting the task scheduler to communicate
// the bound port to the cluster manager properly
// 在启动任务计划程序以将绑定的端口正确通信到集群管理器之前，先绑定UI
_ui.foreach(_.bind())
//默认生成hadoop配置
_hadoopConfiguration = SparkHadoopUtil.get.newConfiguration(_conf)
// Add each JAR given through the constructor
// jar和file添加
if (jars != null) {
jars.foreach(addJar)
}
if (files != null) {
files.foreach(addFile)
}
七、ExecutorMemory配置
代码语言：javascript复制 // executor内存 根据以下属性逐级查找 如果都没有的话最后使用1024MB
_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)
// Convert java options to env vars as a work around
// since we can't set env vars directly in sbt.
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
}
// The Mesos scheduler backend relies on this environment variable to set executor memory.
// Mesos调度程序后端依赖于此环境变量来设置执行程序内存。
// TODO: Set this only in the Mesos scheduler.
executorEnvs("SPARK_EXECUTOR_MEMORY") = executorMemory "m"
executorEnvs = _conf.getExecutorEnv
executorEnvs("SPARK_USER") = sparkUser
八、注册HeartbeatReceiver
心跳接收器，所有 Executor 都会向HeartbeatReceiver 发送心跳，当其接收到 Executor 的心跳信息后，首先更新 Executor 的最后可见时间，然后将此信息交给 TaskScheduler 进一步处理。
代码语言：javascript复制 // We need to register "HeartbeatReceiver" before "createTaskScheduler" because Executor will
// retrieve "HeartbeatReceiver" in the constructor. (SPARK-6640)
// 我们需要在“ createTaskScheduler”之前注册“ HeartbeatReceiver”，
// 因为执行器将在构造函数中检索“ HeartbeatReceiver”。 （SPARK-6640）
_heartbeatReceiver = env.rpcEnv.setupEndpoint(
HeartbeatReceiver.ENDPOINT_NAME, new HeartbeatReceiver(this))
九、创建TaskScheduler
Spark任务调度器，负责任务的提交，并且请求集群管理器对任务调度。由于它调度的Task是有DagScheduler创建，所以DagScheduler是它的前置调度器。
代码语言：javascript复制 val (sched, ts) = SparkContext.createTaskScheduler(this, master, deployMode)
_schedulerBackend = sched
_taskScheduler = ts
十、创建&启动DAGScheduler
一个基于Stage的调度器， 负责创建 Job，将 DAG 中的 RDD 划分到不同的 Stage，并将Stage作为Tasksets提交给底层调度器TaskScheduler执行。
代码语言：javascript复制 //创建DAGScheduler 传入当前SparkContext对象，然后又去取出taskScheduler
// def this(sc: SparkContext) = this(sc, sc.taskScheduler)
_dagScheduler = new DAGScheduler(this)
//绑定心跳执行器
_heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)
// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's constructor
// 在taskScheduler在DAGScheduler的构造函数中设置DAGScheduler引用之后，初始化TaskScheduler
_taskScheduler.start()
_applicationId = _taskScheduler.applicationId()
_applicationAttemptId = taskScheduler.applicationAttemptId()
_conf.set("spark.app.id", _applicationId)
if (_conf.getBoolean("spark.ui.reverseProxy", false)) {
System.setProperty("spark.ui.proxyBase", "/proxy/" _applicationId)
}
_ui.foreach(_.setAppId(_applicationId)) // 啪啪啪一丢设置后 UI和任务关联
十一、初始化BlockManager
在DAGShceduler中有一个BlockManagerMaster对象，该对象的工作就是负责管理全局所有BlockManager的元数据，当集群中有BlockManager注册完成的时候，其会向BlockManagerMaster发送自己元数据信息；BlockManagerMaster会为BlockManager创建一个属于这个BlockManager的BlockManagerInfo，用于存放BlockManager的信息。
代码语言：javascript复制 _env.blockManager.initialize(_applicationId)
十二、初始化MericsSystem
Spark webui 监控指标。包括Shuffle read/wirte gc...etc。spark运行时监控。
代码语言：javascript复制 // The metrics system for Driver need to be set spark.app.id to app ID.
// So it should start after we get app ID from the task scheduler and set spark.app.id.
// 需要将驱动程序的指标系统设置为spark.app.id到应用程序ID。
// 因此，它应该在我们从任务计划程序获取应用程序ID并设置spark.app.id之后开始。
//启动指标监控系统 gc时间，shuffler read/write...etc.
_env.metricsSystem.start()
// Attach the driver metrics servlet handler to the web ui after the metrics system is started.
// 启动指标系统后，将驱动程序指标servlet处理程序附加到Web ui。
_env.metricsSystem.getServletHandlers.foreach(handler => ui.foreach(_.attachHandler(handler)))
十三、创建EventLoggingListener
事件监听器将各种事件进行json转换
代码语言：javascript复制 //创建事件日志监听 添加到总线列队中去（总线列队后面会详细讲~~~）
_eventLogger =
if (isEventLogEnabled) {
val logger =
new EventLoggingListener(_applicationId, _applicationAttemptId, _eventLogDir.get,
_conf, _hadoopConfiguration)
logger.start()
listenerBus.addToEventLogQueue(logger)
Some(logger)
} else {
None
}
十四、创建&启动资源划分管理器
根据配置判断是否开启动态资源管理器
代码语言：javascript复制 // Optionally scale number of executors dynamically based on workload. Exposed for testing.
// 可以根据工作负载动态伸缩执行器的数量spark.dynamicAllocation.enabled
val dynamicAllocationEnabled = Utils.isDynamicAllocationEnabled(_conf)
_executorAllocationManager =
if (dynamicAllocationEnabled) {
schedulerBackend match {
case b: ExecutorAllocationClient =>
Some(new ExecutorAllocationManager(
schedulerBackend.asInstanceOf[ExecutorAllocationClient], listenerBus, _conf,
_env.blockManager.master))
case _ =>
None
}
} else {
None
}
_executorAllocationManager.foreach(_.start())
十五、创建ContextCleaner
上下文清理器，为RDD、shuffle、broadcast状态的异步清理器，清理超出应用范围的RDD、ShuffleDependency、Broadcast对象。
代码语言：javascript复制 //根据spark.cleaner.referenceTracking 默认是true 创建ContextCleaner
_cleaner =
if (_conf.getBoolean("spark.cleaner.referenceTracking", true)) {
Some(new ContextCleaner(this))
} else {
None
}
_cleaner.foreach(_.start())
十六、ExtraListeners配置
代码语言：javascript复制 setupAndStartListenerBus()
十七、环境更新
代码语言：javascript复制postEnvironmentUpdate()
---------------------------------------------------------------
//环境更新，这个就是在UI Environment界面显示的数据
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)
}
}
十八、投递程序启动事件
代码语言：javascript复制postApplicationStart()
--------------------------------------------------------------
//主要关注post方法，发送事件。同样关注start() stop()
private def postApplicationStart() {
// Note: this code assumes that the task scheduler has been initialized and has contacted
// the cluster manager to get an application ID (in case the cluster manager provides one).
listenerBus.post(SparkListenerApplicationStart(appName, Some(applicationId),
startTime, sparkUser, applicationAttemptId, schedulerBackend.getDriverLogUrls))
}
十九、最后绑定注册
代码语言：javascript复制 //关注postStartHook
_taskScheduler.postStartHook()
_env.metricsSystem.registerSource(_dagScheduler.metricsSource)
_env.metricsSystem.registerSource(new BlockManagerSource(_env.blockManager))
_executorAllocationManager.foreach { e =>
_env.metricsSystem.registerSource(e.executorAllocationManagerSource)
二十、结束清理
代码语言：javascript复制 // Make sure the context is stopped if the user forgets about it. This avoids leaving
// unfinished event logs around after the JVM exits cleanly. It doesn't help if the JVM
// is killed, though.
// 如果用户忘记上下文，请确保上下文已停止。这样可以避免在JVM干净退出之后
// 保留未完成的事件日志。但是，如果杀死了JVM 则无济于事
logDebug("Adding shutdown hook")
_shutdownHookRef = ShutdownHookManager.addShutdownHook(
ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY) { () =>
logInfo("Invoking stop() from shutdown hook")
try {
stop()
} catch {
case e: Throwable =>
logWarning("Ignoring Exception while stopping SparkContext from shutdown hook", e)
}
}
} catch {
case NonFatal(e) =>
logError("Error initializing SparkContext.", e)
try {
stop()
} catch {
case NonFatal(inner) =>
logError("Error stopping SparkContext after init error.", inner)
} finally {
throw e
}
}
总结
- Spark 中的组件很多，涉及网络通信、分布式、消息、存储、计算、缓存、测量、清理、文件服务、Web UI 的等等。
- Spark中大量采用事件监听方式，实现driver端的组件之间的通信。
