SparkContext 源码分析

2023-11-07 19:20
文章标签 分析 源码 sparkcontext

本文主要是介绍SparkContext 源码分析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

SparkContext 源码分析

更多资源

  • github: https://github.com/opensourceteams/spark-scala-maven
  • csdn(汇总视频在线看): https://blog.csdn.net/thinktothings/article/details/84726769

Youtub 视频分享

  • Youtub视频(Spark原理分析图解): https://youtu.be/euIuutjAB4I
  • Youtub视频(Spark源码分析详解): https://youtu.be/tUH7QnCcwgg

bilibili 视频分享

  • bilibili视频(Spark原理分析图解): https://youtu.be/euIuutjAB4I
  • bilibili视频(Spark源码分析详解): https://www.bilibili.com/video/av37442161/
src="//player.bilibili.com/player.html?aid=37442161&page=1" scrolling="no" border="0" allowfullscreen="true">

文档说明

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.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.

翻译

).Spark功能主要入口点
).一个SparkContext表示与一个Spark集群的连接
).在Spark集群上,能创建RDDs,累加器,广播变量
).每个JVM仅仅只有一个SparkContext可能是活动的
).在创建一个新的SparkContext之前,你必须停掉活动的SparkContext,这个限制最终可能被 移除,看SPARK-2243 更多详情

SparkContext原理图

SparkContext原理图

xmind文件下载

https://github.com/opensourceteams/spark-scala-maven/blob/master/md/images/spark/SparkContext.xmind

配置信息

可配置信息

  • spark.jars = jar文件路径(可迭代的)
  • spark.files = 文件路径
  • spark.eventLog.dir=/tmp/spark-events // 事件日志目录
  • spark.eventLog.compress=false //事件日志是否压缩
  • spark.shuffle.manager=sort //指定shuffler manager
// Let the user specify short names for shuffle managersval shortShuffleMgrNames = Map("hash" -> "org.apache.spark.shuffle.hash.HashShuffleManager","sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager","tungsten-sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager")val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName)val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)
  • spark.memory.useLegacyMode=true //指定内存管理器
      val useLegacyMemoryManager = conf.getBoolean("spark.memory.useLegacyMode", true)val memoryManager: MemoryManager =if (useLegacyMemoryManager) {new StaticMemoryManager(conf, numUsableCores)} else {UnifiedMemoryManager(conf, numUsableCores)}
    
- spark.ui.showConsoleProgress=true      //展示控制台的进度信息
- spark.ui.enabled=true      //是否开启SparkUI
- spark.executor.memory=      //spark executor 的内存
- SPARK_EXECUTOR_MEMORY=  //spark executor 的内存
- SPARK_MEM=  //spark executor 的内存```scala/***查找顺序,找到前面的就不找后面的了(配置文件中设置值单位为 byte),默认值为1024MBspark.executor.memory  >  SPARK_EXECUTOR_MEMORY >  SPARK_MEM*/_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)
  • spark.scheduler.mode=FIFO //TaskSchedulerImpl 调度模式,可选(FIFO,FAIR,NONE)
/**
*  "FAIR" and "FIFO" determines which policy is used
*    to order tasks amongst a Schedulable's sub-queues
*  "NONE" is used when the a Schedulable has no sub-queues.
*/
object SchedulingMode extends Enumeration {type SchedulingMode = Valueval FAIR, FIFO, NONE = Value
}
  • spark.cores.max=2 设置executor占用cpu内核个数

  • spark.executor.extraJavaOptions= //设置executor启动执行的java参数

  • spark.executor.extraClassPath= //设置 executor 执行的classpath

  • spark.executor.extraLibraryPath= //设置 executor LibraryPath

  • spark.executor.cores= //executor core 个数分配

  • spark.rpc.lookupTimeout=“120s” //设置 RPCTimeout超时时间

  • spark.network.timeout=“120s” //设置 RPCTimeout超时时间

 /** Returns the default Spark timeout to use for RPC remote endpoint lookup. */private[spark] def lookupRpcTimeout(conf: SparkConf): RpcTimeout = {RpcTimeout(conf, Seq("spark.rpc.lookupTimeout", "spark.network.timeout"), "120s")}

Spark系统设置配置信息

  • spark.driver.host = Utils.localHostName()
  • spark.driver.port = 0
  • spark.executor.id = driver

主要内容

创建作业进度监听器

 _jobProgressListener = new JobProgressListener(_conf)listenerBus.addListener(jobProgressListener)

创建SparkEnv

_env = createSparkEnv(_conf, isLocal, listenerBus)
SparkEnv.set(_env)
  • 创建DriverEnv
 SparkEnv.createDriverEnv(conf, isLocal, listenerBus, SparkContext.numDriverCores(master))
  • 指定默认的 spark.rpc = org.apache.spark.rpc.netty.NettyRpcEnvFactory
  private def getRpcEnvFactory(conf: SparkConf): RpcEnvFactory = {val rpcEnvNames = Map("akka" -> "org.apache.spark.rpc.akka.AkkaRpcEnvFactory","netty" -> "org.apache.spark.rpc.netty.NettyRpcEnvFactory")val rpcEnvName = conf.get("spark.rpc", "netty")val rpcEnvFactoryClassName = rpcEnvNames.getOrElse(rpcEnvName.toLowerCase, rpcEnvName)Utils.classForName(rpcEnvFactoryClassName).newInstance().asInstanceOf[RpcEnvFactory]}
  • 创建NettyRpcEnv并启动,此时启动 ‘sparkDriver’
def create(config: RpcEnvConfig): RpcEnv = {val sparkConf = config.conf// Use JavaSerializerInstance in multiple threads is safe. However, if we plan to support// KryoSerializer in future, we have to use ThreadLocal to store SerializerInstanceval javaSerializerInstance =new JavaSerializer(sparkConf).newInstance().asInstanceOf[JavaSerializerInstance]val nettyEnv =new NettyRpcEnv(sparkConf, javaSerializerInstance, config.host, config.securityManager)if (!config.clientMode) {val startNettyRpcEnv: Int => (NettyRpcEnv, Int) = { actualPort =>nettyEnv.startServer(actualPort)(nettyEnv, nettyEnv.address.port)}try {Utils.startServiceOnPort(config.port, startNettyRpcEnv, sparkConf, config.name)._1} catch {case NonFatal(e) =>nettyEnv.shutdown()throw e}}nettyEnv}
  • 创建ActorSystem并启动,此时启动 ‘sparkDriverActorSystem’
 /*** Creates an ActorSystem ready for remoting, with various Spark features. Returns both the* ActorSystem itself and its port (which is hard to get from Akka).** Note: the `name` parameter is important, as even if a client sends a message to right* host + port, if the system name is incorrect, Akka will drop the message.** If indestructible is set to true, the Actor System will continue running in the event* of a fatal exception. This is used by [[org.apache.spark.executor.Executor]].*/def createActorSystem(name: String,host: String,port: Int,conf: SparkConf,securityManager: SecurityManager): (ActorSystem, Int) = {val startService: Int => (ActorSystem, Int) = { actualPort =>doCreateActorSystem(name, host, actualPort, conf, securityManager)}Utils.startServiceOnPort(port, startService, conf, name)}
  • 指定spark序列化器: org.apache.spark.serializer.JavaSerializer

val serializer = instantiateClassFromConfSerializer
logDebug(s"Using serializer: ${serializer.getClass}")

val closureSerializer = instantiateClassFromConf[Serializer]("spark.closure.serializer", "org.apache.spark.serializer.JavaSerializer")

- 实例化 MapOutputTrackerMaster ```scalaval broadcastManager = new BroadcastManager(isDriver, conf, securityManager)val mapOutputTracker = if (isDriver) {new MapOutputTrackerMaster(conf, broadcastManager, isLocal)} else {new MapOutputTrackerWorker(conf)}
  • 注册 MapOutputTracker 到 NettyRpcEndpointRef(通信用 和map输出信息的追踪)
def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {val addr = RpcEndpointAddress(nettyEnv.address, name)val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)synchronized {if (stopped) {throw new IllegalStateException("RpcEnv has been stopped")}if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")}val data = endpoints.get(name)endpointRefs.put(data.endpoint, data.ref)receivers.offer(data)  // for the OnStart message}endpointRef}
  • 实例化ShuflleManager
// Let the user specify short names for shuffle managersval shortShuffleMgrNames = Map("hash" -> "org.apache.spark.shuffle.hash.HashShuffleManager","sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager","tungsten-sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager")val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName)val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)
  • 实例化内存管理器

val useLegacyMemoryManager = conf.getBoolean(“spark.memory.useLegacyMode”, true)
val memoryManager: MemoryManager =
if (useLegacyMemoryManager) {
new StaticMemoryManager(conf, numUsableCores)
} else {
UnifiedMemoryManager(conf, numUsableCores)
}

- 注册 BlockManagerMaster  到 NettyRpcEndpointRef(通信用)```scalaval blockManagerMaster = new BlockManagerMaster(registerOrLookupEndpoint(BlockManagerMaster.DRIVER_ENDPOINT_NAME,new BlockManagerMasterEndpoint(rpcEnv, isLocal, conf, listenerBus)),conf, isDriver)
  • 缓存管理器 实例化BlockManager
   // NB: blockManager is not valid until initialize() is called later.val blockManager = new BlockManager(executorId, rpcEnv, blockManagerMaster,serializer, conf, memoryManager, mapOutputTracker, shuffleManager,blockTransferService, securityManager, numUsableCores)val cacheManager = new CacheManager(blockManager)
  • 创建测量系统
val metricsSystem = if (isDriver) {// Don't start metrics system right now for Driver.// We need to wait for the task scheduler to give us an app ID.// Then we can start the metrics system.MetricsSystem.createMetricsSystem("driver", conf, securityManager)} else {// We need to set the executor ID before the MetricsSystem is created because sources and// sinks specified in the metrics configuration file will want to incorporate this executor's// ID into the metrics they report.conf.set("spark.executor.id", executorId)val ms = MetricsSystem.createMetricsSystem("executor", conf, securityManager)ms.start()ms}
  • 创建临时目录,如果是分布式模式,这是一个executor的当前工作目录
 // Set the sparkFiles directory, used when downloading dependencies.  In local mode,// this is a temporary directory; in distributed mode, this is the executor's current working// directory.val sparkFilesDir: String = if (isDriver) {Utils.createTempDir(Utils.getLocalDir(conf), "userFiles").getAbsolutePath} else {"."}
  • 注册 OutputCommitCoordinator 到 NettyRpcEndpointRef(通信用)
val outputCommitCoordinator = mockOutputCommitCoordinator.getOrElse {new OutputCommitCoordinator(conf, isDriver)}val outputCommitCoordinatorRef = registerOrLookupEndpoint("OutputCommitCoordinator",new OutputCommitCoordinatorEndpoint(rpcEnv, outputCommitCoordinator))outputCommitCoordinator.coordinatorRef = Some(outputCommitCoordinatorRef)
  • new SparkEnv 并返回
 val envInstance = new SparkEnv(executorId,rpcEnv,actorSystem,serializer,closureSerializer,cacheManager,mapOutputTracker,shuffleManager,broadcastManager,blockTransferService,blockManager,securityManager,sparkFilesDir,metricsSystem,memoryManager,outputCommitCoordinator,conf)// Add a reference to tmp dir created by driver, we will delete this tmp dir when stop() is// called, and we only need to do it for driver. Because driver may run as a service, and if we// don't delete this tmp dir when sc is stopped, then will create too many tmp dirs.if (isDriver) {envInstance.driverTmpDirToDelete = Some(sparkFilesDir)}envInstance

创建SparkUI

   _ui =if (conf.getBoolean("spark.ui.enabled", true)) {Some(SparkUI.createLiveUI(this, _conf, listenerBus, _jobProgressListener,_env.securityManager, appName, startTime = startTime))} else {// For tests, do not enable the UINone}// Bind the UI before starting the task scheduler to communicate// the bound port to the cluster manager properly_ui.foreach(_.bind())

注册心跳接收器

   // We need to register "HeartbeatReceiver" before "createTaskScheduler" because Executor will// retrieve "HeartbeatReceiver" in the constructor. (SPARK-6640)_heartbeatReceiver = env.rpcEnv.setupEndpoint(HeartbeatReceiver.ENDPOINT_NAME, new HeartbeatReceiver(this))

创建和启动调度器(TaskScheduler,DAGScheduler)

   // Create and start the schedulerval (sched, ts) = SparkContext.createTaskScheduler(this, master)_schedulerBackend = sched_taskScheduler = ts_dagScheduler = new DAGScheduler(this)_heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's// constructor_taskScheduler.start()
  • org.apache.spark.scheduler.TaskSchedulerImpl 文档说明
/**
).SchedulerBackend 对多种类型的集群调度任务
).LocalBackend 设置  isLocal为true, 也能调度本地任务
).SchedulerBackend.处理常用逻辑,决定跨作业的调度顺序,唤醒和启动推测的任务
).客户端应该先调用  initialize() 和  start(),然后通过 runTasks方法提交任务集* Schedules tasks for multiple types of clusters by acting through a SchedulerBackend.* It can also work with a local setup by using a LocalBackend and setting isLocal to true.* It handles common logic, like determining a scheduling order across jobs, waking up to launch* speculative tasks, etc.** Clients should first call initialize() and start(), then submit task sets through the* runTasks method.** THREADING: SchedulerBackends and task-submitting clients can call this class from multiple* threads, so it needs locks in public API methods to maintain its state. In addition, some* SchedulerBackends synchronize on themselves when they want to send events here, and then* acquire a lock on us, so we need to make sure that we don't try to lock the backend while* we are holding a lock on ourselves.*/
  • Standalone模式创建TaskSchedulerImpl并初使化中指定 backend为SparkDeploySchedulerBackend
     case SPARK_REGEX(sparkUrl) =>val scheduler = new TaskSchedulerImpl(sc)val masterUrls = sparkUrl.split(",").map("spark://" + _)val backend = new SparkDeploySchedulerBackend(scheduler, sc, masterUrls)scheduler.initialize(backend)(backend, scheduler)
 def initialize(backend: SchedulerBackend) {this.backend = backend// temporarily set rootPool name to emptyrootPool = new Pool("", schedulingMode, 0, 0)schedulableBuilder = {schedulingMode match {case SchedulingMode.FIFO =>new FIFOSchedulableBuilder(rootPool)case SchedulingMode.FAIR =>new FairSchedulableBuilder(rootPool, conf)}}schedulableBuilder.buildPools()}

任务调度器启动

  • 任务调度器启动_taskScheduler.start()
// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's// constructor_taskScheduler.start()
  • 调用SparkDeploySchedulerBackend start方法
  override def start() {backend.start()if (!isLocal && conf.getBoolean("spark.speculation", false)) {logInfo("Starting speculative execution thread")speculationScheduler.scheduleAtFixedRate(new Runnable {override def run(): Unit = Utils.tryOrStopSparkContext(sc) {checkSpeculatableTasks()}}, SPECULATION_INTERVAL_MS, SPECULATION_INTERVAL_MS, TimeUnit.MILLISECONDS)}}
  • 再调用CoarseGrainedSchedulerBackend 的start方法 registerRpcEndpoint
    注册(通信用) [CoarseGrainedScheduler]
  • 实例化ApplicationDescription 包含 command (org.apache.spark.executor.CoarseGrainedExecutorBackend)
  • 启动 AppClient registerRpcEndpoint 注册(通信用)[AppClient]
override def start() {super.start()launcherBackend.connect()// The endpoint for executors to talk to usval driverUrl = rpcEnv.uriOf(SparkEnv.driverActorSystemName,RpcAddress(sc.conf.get("spark.driver.host"), sc.conf.get("spark.driver.port").toInt),CoarseGrainedSchedulerBackend.ENDPOINT_NAME)val args = Seq("--driver-url", driverUrl,"--executor-id", "{{EXECUTOR_ID}}","--hostname", "{{HOSTNAME}}","--cores", "{{CORES}}","--app-id", "{{APP_ID}}","--worker-url", "{{WORKER_URL}}")val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions").map(Utils.splitCommandString).getOrElse(Seq.empty)val classPathEntries = sc.conf.getOption("spark.executor.extraClassPath").map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)val libraryPathEntries = sc.conf.getOption("spark.executor.extraLibraryPath").map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)// When testing, expose the parent class path to the child. This is processed by// compute-classpath.{cmd,sh} and makes all needed jars available to child processes// when the assembly is built with the "*-provided" profiles enabled.val testingClassPath =if (sys.props.contains("spark.testing")) {sys.props("java.class.path").split(java.io.File.pathSeparator).toSeq} else {Nil}// Start executors with a few necessary configs for registering with the schedulerval sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)val javaOpts = sparkJavaOpts ++ extraJavaOptsval command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)val appUIAddress = sc.ui.map(_.appUIAddress).getOrElse("")val coresPerExecutor = conf.getOption("spark.executor.cores").map(_.toInt)val appDesc = new ApplicationDescription(sc.appName, maxCores, sc.executorMemory,command, appUIAddress, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor)client = new AppClient(sc.env.rpcEnv, masters, appDesc, this, conf)client.start()launcherBackend.setState(SparkAppHandle.State.SUBMITTED)waitForRegistration()launcherBackend.setState(SparkAppHandle.State.RUNNING)}
  • CoarseGrainedSchedulerBackend 文档
/**
). 一个后端调度器 等待粗粒度 executors 通过 Akka连接他
).在Spark作业期间,这个后端调度接管着每个executor 比起交出executors给别人调度
).无论何时完成任务,都要求调度器启动一个新的executor给每个每个新的任务
).executors 可以以多种方式启动,例如 粗粒度 Mesos模式 Mesos任务
).或Spark standalone 部署模式的 standalone 处理* A scheduler backend that waits for coarse grained executors to connect to it through Akka.
* This backend holds onto each executor for the duration of the Spark job rather than relinquishing
* executors whenever a task is done and asking the scheduler to launch a new executor for
* each new task. Executors may be launched in a variety of ways, such as Mesos tasks for the
* coarse-grained Mesos mode or standalone processes for Spark's standalone deploy mode
* (spark.deploy.*).
*/
private[spark]
class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv)extends ExecutorAllocationClient with SchedulerBackend with Logging
{
  • 调用ClientEndpoint 的 onStart方法 异步向所有master注册,向master发送消息: RegisterApplication
override def onStart(): Unit = {try {registerWithMaster(1)} catch {case e: Exception =>logWarning("Failed to connect to master", e)markDisconnected()stop()}}
 /*** Register with all masters asynchronously. It will call `registerWithMaster` every* REGISTRATION_TIMEOUT_SECONDS seconds until exceeding REGISTRATION_RETRIES times.* Once we connect to a master successfully, all scheduling work and Futures will be cancelled.** nthRetry means this is the nth attempt to register with master.*/private def registerWithMaster(nthRetry: Int) {registerMasterFutures.set(tryRegisterAllMasters())registrationRetryTimer.set(registrationRetryThread.scheduleAtFixedRate(new Runnable {override def run(): Unit = {if (registered.get) {registerMasterFutures.get.foreach(_.cancel(true))registerMasterThreadPool.shutdownNow()} else if (nthRetry >= REGISTRATION_RETRIES) {markDead("All masters are unresponsive! Giving up.")} else {registerMasterFutures.get.foreach(_.cancel(true))registerWithMaster(nthRetry + 1)}}}, REGISTRATION_TIMEOUT_SECONDS, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))}
   /***  Register with all masters asynchronously and returns an array `Future`s for cancellation.*/private def tryRegisterAllMasters(): Array[JFuture[_]] = {for (masterAddress <- masterRpcAddresses) yield {registerMasterThreadPool.submit(new Runnable {override def run(): Unit = try {if (registered.get) {return}logInfo("Connecting to master " + masterAddress.toSparkURL + "...")val masterRef =rpcEnv.setupEndpointRef(Master.SYSTEM_NAME, masterAddress, Master.ENDPOINT_NAME)masterRef.send(RegisterApplication(appDescription, self))} catch {case ie: InterruptedException => // Cancelledcase NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)}})}}

org.apache.spark.rpc.netty.Dispatcher

类文档说明
/**
一个消息分配器,负责将RPC消息路由到适当的端点
* A message dispatcher, responsible for routing RPC messages to the appropriate endpoint(s).
*/
private[netty] class Dispatcher(nettyEnv: NettyRpcEnv) extends Logging {
注册RPC端点 (关键通信及OnStart方法的调用)
/*** new EndpointData(name, endpoint, endpointRef) 的时候会进行 *  val inbox = new Inbox(ref, endpoint)的操作* Inbox 实例化时会进行如下操作,当相于首先增加OnStart消息// OnStart should be the first message to processinbox.synchronized {messages.add(OnStart)}*/
def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {val addr = RpcEndpointAddress(nettyEnv.address, name)val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)synchronized {if (stopped) {throw new IllegalStateException("RpcEnv has been stopped")}if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")}val data = endpoints.get(name)endpointRefs.put(data.endpoint, data.ref)receivers.offer(data)  // for the OnStart message}endpointRef}
Inbox

A inbox that stores messages for an [[RpcEndpoint]] and posts messages to it thread-safely.

/**
* Inbox 实例化时会进行如下操作,当相于首先增加OnStart消息给当前的对象
(也就是每个RpcEndpoint 子类进行注册时,首先增加OnStart消息)
* OnStart消息在 process方法中会进行 endpoint实现类的onStart() 方法回调
*/// OnStart should be the first message to processinbox.synchronized {messages.add(OnStart)}
  /*** Process stored messages.*/def process(dispatcher: Dispatcher): Unit = {......
case OnStart =>endpoint.onStart()if (!endpoint.isInstanceOf[ThreadSafeRpcEndpoint]) {inbox.synchronized {if (!stopped) {enableConcurrent = true}}}

入口代码块 400行

try {_conf = config.clone()_conf.validateSettings()if (!_conf.contains("spark.master")) {throw new SparkException("A master URL must be set in your configuration")}if (!_conf.contains("spark.app.name")) {throw new SparkException("An application name must be set in your configuration")}// System property spark.yarn.app.id must be set if user code ran by AM on a YARN cluster// yarn-standalone is deprecated, but still supportedif ((master == "yarn-cluster" || master == "yarn-standalone") &&!_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_conf.setIfMissing("spark.driver.host", Utils.localHostName())_conf.setIfMissing("spark.driver.port", "0")_conf.set("spark.executor.id", SparkContext.DRIVER_IDENTIFIER)_jars = _conf.getOption("spark.jars").map(_.split(",")).map(_.filter(_.size != 0)).toSeq.flatten_files = _conf.getOption("spark.files").map(_.split(",")).map(_.filter(_.size != 0)).toSeq.flatten_eventLogDir =if (isEventLogEnabled) {val unresolvedDir = conf.get("spark.eventLog.dir", EventLoggingListener.DEFAULT_LOG_DIR).stripSuffix("/")Some(Utils.resolveURI(unresolvedDir))} else {None}_eventLogCodec = {val compress = _conf.getBoolean("spark.eventLog.compress", false)if (compress && isEventLogEnabled) {Some(CompressionCodec.getCodecName(_conf)).map(CompressionCodec.getShortName)} else {None}}_conf.set("spark.externalBlockStore.folderName", externalBlockStoreFolderName)if (master == "yarn-client") System.setProperty("SPARK_YARN_MODE", "true")// "_jobProgressListener" should be set up before creating SparkEnv because when creating// "SparkEnv", some messages will be posted to "listenerBus" and we should not miss them._jobProgressListener = new JobProgressListener(_conf)listenerBus.addListener(jobProgressListener)// Create the Spark execution environment (cache, map output tracker, etc)_env = createSparkEnv(_conf, isLocal, listenerBus)SparkEnv.set(_env)// If running the REPL, register the repl's output dir with the file server._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)}_metadataCleaner = new MetadataCleaner(MetadataCleanerType.SPARK_CONTEXT, this.cleanup, _conf)_statusTracker = new SparkStatusTracker(this)_progressBar =if (_conf.getBoolean("spark.ui.showConsoleProgress", true) && !log.isInfoEnabled) {Some(new ConsoleProgressBar(this))} else {None}_ui =if (conf.getBoolean("spark.ui.enabled", true)) {Some(SparkUI.createLiveUI(this, _conf, listenerBus, _jobProgressListener,_env.securityManager, appName, startTime = startTime))} else {// For tests, do not enable the UINone}// Bind the UI before starting the task scheduler to communicate// the bound port to the cluster manager properly_ui.foreach(_.bind())_hadoopConfiguration = SparkHadoopUtil.get.newConfiguration(_conf)// Add each JAR given through the constructorif (jars != null) {jars.foreach(addJar)}if (files != null) {files.foreach(addFile)}_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.// TODO: Set this only in the Mesos scheduler.executorEnvs("SPARK_EXECUTOR_MEMORY") = executorMemory + "m"executorEnvs ++= _conf.getExecutorEnvexecutorEnvs("SPARK_USER") = sparkUser// We need to register "HeartbeatReceiver" before "createTaskScheduler" because Executor will// retrieve "HeartbeatReceiver" in the constructor. (SPARK-6640)_heartbeatReceiver = env.rpcEnv.setupEndpoint(HeartbeatReceiver.ENDPOINT_NAME, new HeartbeatReceiver(this))// Create and start the schedulerval (sched, ts) = SparkContext.createTaskScheduler(this, master)_schedulerBackend = sched_taskScheduler = ts_dagScheduler = new DAGScheduler(this)_heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's// constructor_taskScheduler.start()_applicationId = _taskScheduler.applicationId()_applicationAttemptId = taskScheduler.applicationAttemptId()_conf.set("spark.app.id", _applicationId)_ui.foreach(_.setAppId(_applicationId))_env.blockManager.initialize(_applicationId)// 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.metricsSystem.start()// Attach the driver metrics servlet handler to the web ui after the metrics system is started.metricsSystem.getServletHandlers.foreach(handler => ui.foreach(_.attachHandler(handler)))_eventLogger =if (isEventLogEnabled) {val logger =new EventLoggingListener(_applicationId, _applicationAttemptId, _eventLogDir.get,_conf, _hadoopConfiguration)logger.start()listenerBus.addListener(logger)Some(logger)} else {None}// Optionally scale number of executors dynamically based on workload. Exposed for testing.val dynamicAllocationEnabled = Utils.isDynamicAllocationEnabled(_conf)if (!dynamicAllocationEnabled && _conf.getBoolean("spark.dynamicAllocation.enabled", false)) {logWarning("Dynamic Allocation and num executors both set, thus dynamic allocation disabled.")}_executorAllocationManager =if (dynamicAllocationEnabled) {Some(new ExecutorAllocationManager(this, listenerBus, _conf))} else {None}_executorAllocationManager.foreach(_.start())_cleaner =if (_conf.getBoolean("spark.cleaner.referenceTracking", true)) {Some(new ContextCleaner(this))} else {None}_cleaner.foreach(_.start())setupAndStartListenerBus()postEnvironmentUpdate()postApplicationStart()// Post init_taskScheduler.postStartHook()_env.metricsSystem.registerSource(_dagScheduler.metricsSource)_env.metricsSystem.registerSource(new BlockManagerSource(_env.blockManager))_executorAllocationManager.foreach { e =>_env.metricsSystem.registerSource(e.executorAllocationManagerSource)}// 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._shutdownHookRef = ShutdownHookManager.addShutdownHook(ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY) { () =>logInfo("Invoking stop() from shutdown hook")stop()}} 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}}

这篇关于SparkContext 源码分析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》:本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》:本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景

python中Hash使用场景分析

《python中Hash使用场景分析》Python的hash()函数用于获取对象哈希值,常用于字典和集合,不可变类型可哈希,可变类型不可,常见算法包括除法、乘法、平方取中和随机数哈希,各有优缺点,需根... 目录python中的 Hash除法哈希算法乘法哈希算法平方取中法随机数哈希算法小结在Python中,

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重

关于MyISAM和InnoDB对比分析

《关于MyISAM和InnoDB对比分析》:本文主要介绍关于MyISAM和InnoDB对比分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录开篇:从交通规则看存储引擎选择理解存储引擎的基本概念技术原理对比1. 事务支持:ACID的守护者2. 锁机制:并发控制的艺

MyBatis Plus 中 update_time 字段自动填充失效的原因分析及解决方案(最新整理)

《MyBatisPlus中update_time字段自动填充失效的原因分析及解决方案(最新整理)》在使用MyBatisPlus时,通常我们会在数据库表中设置create_time和update... 目录前言一、问题现象二、原因分析三、总结:常见原因与解决方法对照表四、推荐写法前言在使用 MyBATis

Python主动抛出异常的各种用法和场景分析

《Python主动抛出异常的各种用法和场景分析》在Python中,我们不仅可以捕获和处理异常,还可以主动抛出异常,也就是以类的方式自定义错误的类型和提示信息,这在编程中非常有用,下面我将详细解释主动抛... 目录一、为什么要主动抛出异常?二、基本语法:raise关键字基本示例三、raise的多种用法1. 抛

github打不开的问题分析及解决

《github打不开的问题分析及解决》:本文主要介绍github打不开的问题分析及解决,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、找到github.com域名解析的ip地址二、找到github.global.ssl.fastly.net网址解析的ip地址三

Mysql的主从同步/复制的原理分析

《Mysql的主从同步/复制的原理分析》:本文主要介绍Mysql的主从同步/复制的原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录为什么要主从同步?mysql主从同步架构有哪些?Mysql主从复制的原理/整体流程级联复制架构为什么好?Mysql主从复制注意

java -jar命令运行 jar包时运行外部依赖jar包的场景分析

《java-jar命令运行jar包时运行外部依赖jar包的场景分析》:本文主要介绍java-jar命令运行jar包时运行外部依赖jar包的场景分析,本文给大家介绍的非常详细,对大家的学习或工作... 目录Java -jar命令运行 jar包时如何运行外部依赖jar包场景:解决:方法一、启动参数添加: -Xb