0
0

Spark-driver和executor启动过程

一、上下文

《Spark-SparkSubmit详细过程》详细分析了从脚本提交任务后driver是如何调用到自己编写的Spark代码的，而我们的Spark代码在运行前必须准备好分布式资源，接下来我们就分析下资源是如何分配的

二、Spark代码示例

我们以一个简单的WordCount程序为例，来分析Spark后端是如何为这个程序分配资源的

  1. object WordCount {
  2.   def main(args: Array[String]): Unit = {
  3.     //可以通过 SparkConf 为 Spark 绝大多数配置设置参数，且这些参数的优先级要高于系统属性
  4.     //注意：一旦 SparkConf 传递给 Spark 后，就无法再对其进行修改，因为Spark不支持运行时修改
  5.     val conf = new SparkConf().setAppName("WordCount")
  6.     //Spark 的主要入口点 SparkContext 表示到Spark集群的连接，用于在该集群上创建RDD、累加器、广播变量
  7.     //每个JVM只能有一个 SparkContext 处于活动状态
  8.     val sc = new SparkContext(conf)
  9.     //从HDFS、本地文件系统（在所有节点上都可用）或任何Hadoop支持的文件系统URI读取文本文件，并将其作为字符串的RDD返回。
  10.     val sourceRdd = sc.textFile("file/word_count_data.txt")
  11.     //原始一行数据：js,c,vba,json,xml
  12.     //flatMap将每行数据按照逗号分割，得到每个单词 形成 (单词1) (单词2) (单词1) ... 的格式
  13.     //map将每个单词的次数都赋值成1形成 (单词1,1) (单词2,1) (单词1,次数) ... 的格式
  14.     //reduceByKey将相同单词中的次数进行累加
  15.     val resultRdd = sourceRdd.flatMap(_.split(",")).map(x=>{(x,1)}).reduceByKey(_+_)
  16.     //打印结果
  17.     resultRdd.foreach(println)
  18.     //停止SparkContext
  19.     sc.stop()
  21.   }

SparkConf 是对该程序的一个属性设置，且支持链式设置，会覆盖默认的系统属性。一旦程序开始运行就不可以对其再修改了。

SparkContext是程序与Spark集群连接的入口，可以用于在该集群上创建RDD、广播变量、累加器等等，那么RDD运行所需的资源肯定是在创建SparkContext时就已经具备好的。下面我们看看SparkContext中是如何结合spark-submit参数来协调资源的

三、SparkContext

  1. class SparkContext(config: SparkConf) extends Logging {
  3.   //创建一个从系统属性加载设置的SparkContext（例如，使用./bin/spark-submit启动时）
  4.   def this() = this(new SparkConf())
  6.   //......省略.......
  8.   //私有变量。这些变量保留了上下文的内部状态，外部世界无法访问。它们是可变的
  9.   //因为我们想提前将它们初始化为一个中性值，这样在构造函数仍在运行时调用“stop（）”是安全的。
  10.   //只列举重要的一些属性
  11.   //支持应用的个性化配置
  12.   private var _conf: SparkConf = _
  13.   //一个正在运行的Spark实例的所有运行时环境对象（无论是master还是worker），包括序列化器、RpcEnv、块管理器、映射输出跟踪器等。目前，Spark代码通过全局变量查找SparkEnv，因此所有线程都可以访问相同的SparkEnv。它可以通过SparkEnv.get访问（例如在创建SparkContext之后）
  14.   //我们后面单独分析下它
  15.   private var _env: SparkEnv = _
  16.   //用于调度系统的后端接口，允许在TaskSchedulerImpl下插入不同的系统。我们假设一个类似Mesos的模型，当机器可用时，应用程序会获得资源供应，并可以在机器上启动任务。
  17.   private var _schedulerBackend: SchedulerBackend = _
  18.   //低级任务调度程序接口，目前由[[org.apache.sspark.scheduler.TaskSchedulerImpl]]专门实现。 
  19.   //此接口允许插入不同的任务调度器。每个TaskScheduler都为单个SparkContext安排任务。
  20.   //这些调度器从DAGScheduler获取每个阶段提交给它们的任务集，并负责将任务发送到集群、运行它们、在出现故障时重试，以及缓解延迟。他们将事件返回给DAGScheduler。
  21.   private var _taskScheduler: TaskScheduler = _
  22.   //一个常住在 Driver端的 HeartbeatReceiver 通信端点 ，用来接收所有 executors 的心跳
  23.   private var _heartbeatReceiver: RpcEndpointRef = _
  24.   //实现面向 stage 调度的高级调度层。它为每个作业计算一个 stage 的DAG，
  25.   //跟踪哪些RDD和stage输出被物化，并找到运行作业的最小时间表。
  26.   //然后，它将stage作为TaskSet提交给在集群上运行它们的底层TaskScheduler实现。
  27.   //TaskSet包含完全独立的任务，这些任务可以根据集群上已有的数据
  28.   @volatile private var _dagScheduler: DAGScheduler = _
  30.   try {
  31.     //......省略.......
  32.     //创建SparkEnv
  33.     _env = createSparkEnv(_conf, isLocal, listenerBus)
  34.     //设置 常住driver端的 _heartbeatReceiver 的 Endpoint
  35.     _heartbeatReceiver = env.rpcEnv.setupEndpoint(
  36.       HeartbeatReceiver.ENDPOINT_NAME, new HeartbeatReceiver(this))
  38.     // 创建并启动调度程序
  39.     val (sched, ts) = SparkContext.createTaskScheduler(this, master, deployMode)
  40.     _schedulerBackend = sched
  41.     _taskScheduler = ts
  42.     _dagScheduler = new DAGScheduler(this)
  43.     _heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)
  45.     //启动任务调度器
  46.     _taskScheduler.start()
  48.     //......省略.......  
  50.   //基于给定的主URL创建任务调度器。返回调度器后端和任务调度器的 2-tuple
  51.   private def createTaskScheduler(...){
  52.     //......省略.......  
  53.     master match {
  54.       case "local" =>
  55.       case LOCAL_N_REGEX(threads) =>
  56.       case LOCAL_N_FAILURES_REGEX(threads, maxFailures) =>
  57.       case SPARK_REGEX(sparkUrl) =>
  58.         //创建一个 任务调度实现类 TaskSchedulerImpl
  59.         val scheduler = new TaskSchedulerImpl(sc)
  60.         val masterUrls = sparkUrl.split(",").map("spark://" + _)
  61.         //创建一个 StandaloneSchedulerBackend 是 SchedulerBackend 的子类
  62.         val backend = new StandaloneSchedulerBackend(scheduler, sc, masterUrls)
  63.         //初始化 任务调度器 
  64.         scheduler.initialize(backend)
  65.         (backend, scheduler)
  66.       case LOCAL_CLUSTER_REGEX(numWorkers, coresPerWorker, memoryPerWorker) =>
  67.       case masterUrl =>
  68.   }
  70. }

四、TaskSchedulerImpl

  1. private[spark] class TaskSchedulerImpl(
  2.     val sc: SparkContext,
  3.     val maxTaskFailures: Int,
  4.     isLocal: Boolean = false,
  5.     clock: Clock = new SystemClock)
  6.   extends TaskScheduler with Logging {
  8.   //初始化
  9.   def initialize(backend: SchedulerBackend): Unit = {
  10.     this.backend = backend
  11.     //构建调度器   默认是 FIFO 调度 可以通过 spark.scheduler.mode 进行配置
  12.     schedulableBuilder = {
  13.       schedulingMode match {
  14.         case SchedulingMode.FIFO =>
  15.           new FIFOSchedulableBuilder(rootPool)
  16.         case SchedulingMode.FAIR =>
  17.           new FairSchedulableBuilder(rootPool, sc)
  18.         case _ =>
  19.           throw new IllegalArgumentException(s"Unsupported $SCHEDULER_MODE_PROPERTY: " +
  20.           s"$schedulingMode")
  21.       }
  22.     }
  23.     schedulableBuilder.buildPools()
  24.   }
  26.   //启动调度程序
  27.   override def start(): Unit = {
  28.     //SparkContext中创建的是 StandaloneSchedulerBackend 因此会调用 它的 start()
  29.     //StandaloneSchedulerBackend  又会调用其父类CoarseGrainedSchedulerBackend 的 start() 
  30.     backend.start()
  32.     if (!isLocal && conf.get(SPECULATION_ENABLED)) {
  33.       logInfo("Starting speculative execution thread")
  34.       speculationScheduler.scheduleWithFixedDelay(
  35.         () => Utils.tryOrStopSparkContext(sc) { checkSpeculatableTasks() },
  36.         SPECULATION_INTERVAL_MS, SPECULATION_INTERVAL_MS, TimeUnit.MILLISECONDS)
  37.     }
  38.   }
  39. }

五、CoarseGrainedSchedulerBackend

  1. //等待粗粒度执行器连接的调度程序后端。此后端在Spark作业期间保留每个执行器，而不是在任务完成时放弃执行器，
  2. //并要求调度器为每个新任务启动一个新的执行器。执行器可以通过多种方式启动，例如粗粒度Mesos模式的Mesos任务或Spark独立部署模式（Spark.deploy.*）的独立进程。
  3. class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv)
  4.   extends ExecutorAllocationClient with SchedulerBackend with Logging {
  6.   //这里会创建并注册一个 DriverEndpoint ，且 DriverEndpoint的 onStart() 方法会执行
  7.   val driverEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint())
  8.   protected def createDriverEndpoint(): DriverEndpoint = new DriverEndpoint()
  10.   override def start(): Unit = {
  12.   }
  14. class DriverEndpoint extends IsolatedRpcEndpoint with Logging {
  15.     override def onStart(): Unit = {
  16.       // 定期恢复下 以允许延迟调度工作
  17.       val reviveIntervalMs = conf.get(SCHEDULER_REVIVE_INTERVAL).getOrElse(1000L)
  19.       reviveThread.scheduleAtFixedRate(() => Utils.tryLogNonFatalError {
  20.         Option(self).foreach(_.send(ReviveOffers))
  21.       }, 0, reviveIntervalMs, TimeUnit.MILLISECONDS)
  22.     }
  24.     override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
  25.       case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls,
  26.           attributes, resources, resourceProfileId) =>
  27.         if (executorDataMap.contains(executorId)) {
  28.           context.sendFailure(new IllegalStateException(s"Duplicate executor ID: $executorId"))
  29.         } else if (scheduler.excludedNodes.contains(hostname) ||
  30.             isExecutorExcluded(executorId, hostname)) {
  31.           // 如果集群管理器在被排除的节点上为我们提供了一个Executor（因为在我们通知它我们的排除之前，它已经开始分配这些资源，或者如果它忽略了我们的排除），那么我们会立即拒绝该Executor
  32.           logInfo(s"Rejecting $executorId as it has been excluded.")
  33.           context.sendFailure(
  34.             new IllegalStateException(s"Executor is excluded due to failures: $executorId"))
  35.         } else {
  36.           //如果Executor 的rpc-env没有监听传入连接，则“hostPort”将为null，应使用客户端连接联系Executor 。
  37.           val executorAddress = if (executorRef.address != null) {
  38.               executorRef.address
  39.             } else {
  40.               context.senderAddress
  41.             }
  42.           logInfo(s"Registered executor $executorRef ($executorAddress) with ID $executorId, " +
  43.             s" ResourceProfileId $resourceProfileId")
  44.           addressToExecutorId(executorAddress) = executorId
  45.           totalCoreCount.addAndGet(cores)
  46.           totalRegisteredExecutors.addAndGet(1)
  47.           val resourcesInfo = resources.map { case (rName, info) =>
  48.             // 这必须同步，因为在请求Executor时会读取此块中突变的变量
  49.             val numParts = scheduler.sc.resourceProfileManager
  50.               .resourceProfileFromId(resourceProfileId).getNumSlotsPerAddress(rName, conf)
  51.             (info.name, new ExecutorResourceInfo(info.name, info.addresses, numParts))
  52.           }
  53.           val data = new ExecutorData(executorRef, executorAddress, hostname,
  54.             0, cores, logUrlHandler.applyPattern(logUrls, attributes), attributes,
  55.             resourcesInfo, resourceProfileId, registrationTs = System.currentTimeMillis())
  56.           // This must be synchronized because variables mutated
  57.           // in this block are read when requesting executors
  58.           CoarseGrainedSchedulerBackend.this.synchronized {
  59.             executorDataMap.put(executorId, data)
  60.             if (currentExecutorIdCounter < executorId.toInt) {
  61.               currentExecutorIdCounter = executorId.toInt
  62.             }
  63.           }
  64.           listenerBus.post(
  65.             SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))
  66.           // Note: some tests expect the reply to come after we put the executor in the map
  67.           context.reply(true)
  68.         }
  70.     }
  72.     override def receive: PartialFunction[Any, Unit] = {
  73.       //启动 executor 
  74.       case LaunchedExecutor(executorId) =>
  75.         executorDataMap.get(executorId).foreach { data =>
  76.           data.freeCores = data.totalCores
  77.         }
  78.         //仅仅为一个 executor 提供虚假资源 offer 
  79.         makeOffers(executorId)
  80.     }
  82. }
  84. }

六、StandaloneSchedulerBackend

  1. //Spark独立集群管理器的[[SchedulerBackend]]实现。
  2. private[spark] class StandaloneSchedulerBackend(
  3.     scheduler: TaskSchedulerImpl,
  4.     sc: SparkContext,
  5.     masters: Array[String])
  6.   extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv)
  7.   with StandaloneAppClientListener
  8.   with Logging {
  10.  override def start(): Unit = {
  11.     super.start()
  13.     //调度器后端应仅在客户端模式下尝试连接到启动器。
  14.     //在集群模式下，将应用程序提交给Master的代码需要连接到启动器。
  15.     if (sc.deployMode == "client") {
  16.       launcherBackend.connect()
  17.     }
  19.     //executors 中的 endpoint 需要持有 driver的地址 用于和 driver通信
  20.     val driverUrl = RpcEndpointAddress(
  21.       sc.conf.get(config.DRIVER_HOST_ADDRESS),
  22.       sc.conf.get(config.DRIVER_PORT),
  23.       CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
  24.     val args = Seq(
  25.       "--driver-url", driverUrl,
  26.       "--executor-id", "{{EXECUTOR_ID}}",
  27.       "--hostname", "{{HOSTNAME}}",
  28.       "--cores", "{{CORES}}",
  29.       "--app-id", "{{APP_ID}}",
  30.       "--worker-url", "{{WORKER_URL}}")
  31.     val extraJavaOpts = sc.conf.get(config.EXECUTOR_JAVA_OPTIONS)
  32.       .map(Utils.splitCommandString).getOrElse(Seq.empty)
  33.     val classPathEntries = sc.conf.get(config.EXECUTOR_CLASS_PATH)
  34.       .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
  35.     val libraryPathEntries = sc.conf.get(config.EXECUTOR_LIBRARY_PATH)
  36.       .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
  38.     //使用一些必要的配置启动 executors ，以便在调度程序中注册
  39.     val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)
  40.     val javaOpts = sparkJavaOpts ++ extraJavaOpts
  41.     // executors 端的主类 CoarseGrainedExecutorBackend
  42.     val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",
  43.       args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)
  44.     val webUrl = sc.ui.map(_.webUrl).getOrElse("")
  45.     val coresPerExecutor = conf.getOption(config.EXECUTOR_CORES.key).map(_.toInt)
  46.     // 如果我们使用动态分配，现在将初始执行器限制设置为0。ExecutorDallocationManager稍后会将实际的初始限额发送给Master。
  47.     val initialExecutorLimit =
  48.       if (Utils.isDynamicAllocationEnabled(conf)) {
  49.         Some(0)
  50.       } else {
  51.         None
  52.       }
  53.     val executorResourceReqs = ResourceUtils.parseResourceRequirements(conf,
  54.       config.SPARK_EXECUTOR_PREFIX)
  55.     //这里有一个 ApplicationDescription 之前有要给 DriverDescription
  56.     //可以想到 它是用来启动一个 Application 用的
  57.     val appDesc = ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
  58.       webUrl, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, initialExecutorLimit,
  59.       resourceReqsPerExecutor = executorResourceReqs)
  60.     //创建一个 StandaloneAppClient
  61.     //允许应用程序与Spark独立集群管理器对话。获取集群事件的主URL、应用程序描述和侦听器，并在发生各种事件时回调侦听器。
  62.     client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
  63.     //启动它
  64.     client.start()
  65.     launcherBackend.setState(SparkAppHandle.State.SUBMITTED)
  66.     waitForRegistration()
  67.     launcherBackend.setState(SparkAppHandle.State.RUNNING)
  68.   }
  70. }

七、StandaloneAppClient

  1. private[spark] class StandaloneAppClient(...){
  3.   private val masterRpcAddresses = masterUrls.map(RpcAddress.fromSparkURL(_))
  4.   private val REGISTRATION_TIMEOUT_SECONDS = 20
  5.   private val REGISTRATION_RETRIES = 3
  7.   private class ClientEndpoint(override val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint
  8.     with Logging {
  10.     override def onStart(): Unit = {
  11.         //向Master注册一个app
  12.         registerWithMaster(1)
  13.     }
  15.     //异步向所有 Master 注册。它将每隔 20 秒调用“registerWithMaster”，直到超过 3 次数。一旦我们成功连接到主机，所有调度工作和期货都将被取消。
  16.     //thRetry表示这是第n次尝试向master注册。
  17.     private def registerWithMaster(nthRetry: Int): Unit = {
  18.       registerMasterFutures.set(tryRegisterAllMasters())
  19.       registrationRetryTimer.set(registrationRetryThread.schedule(new Runnable {
  20.         override def run(): Unit = {
  21.           if (registered.get) {
  22.             registerMasterFutures.get.foreach(_.cancel(true))
  23.             registerMasterThreadPool.shutdownNow()
  24.           } else if (nthRetry >= REGISTRATION_RETRIES) {
  25.             markDead("All masters are unresponsive! Giving up.")
  26.           } else {
  27.             registerMasterFutures.get.foreach(_.cancel(true))
  28.             registerWithMaster(nthRetry + 1)
  29.           }
  30.         }
  31.       }, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))
  32.     }
  34.   }
  36.     private def tryRegisterAllMasters(): Array[JFuture[_]] = {
  37.       for (masterAddress <- masterRpcAddresses) yield {
  38.         registerMasterThreadPool.submit(new Runnable {
  39.           override def run(): Unit = try {
  40.             if (registered.get) {
  41.               return
  42.             }
  43.             logInfo("Connecting to master " + masterAddress.toSparkURL + "...")
  44.             val masterRef = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
  45.             //向Master 发送 RegisterApplication 消息
  46.             masterRef.send(RegisterApplication(appDescription, self))
  47.           } catch {
  48.             case ie: InterruptedException => // Cancelled
  49.             case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
  50.           }
  51.         })
  52.       }
  53.     }
  55.   def start(): Unit = {
  56.     // 只需启动一个rpcEndpoint；它将呼叫回听众
  57.     //设置一个 AppClient 端点 ClientEndpoint 的 onstart 的方法会调起
  58.     endpoint.set(rpcEnv.setupEndpoint("AppClient", new ClientEndpoint(rpcEnv)))
  59.   }
  61. }

八、Master

  1. private[deploy] class Master(...){
  3.   //处理其他端点的消息
  4.   override def receive: PartialFunction[Any, Unit] = {
  5.     case RegisterApplication(description, driver) =>
  6.       // TODO Prevent repeated registrations from some driver
  7.       if (state == RecoveryState.STANDBY) {
  8.         // ignore, don't send response
  9.       } else {
  10.         logInfo("Registering app " + description.name)
  11.         //创建一个 app
  12.         val app = createApplication(description, driver)
  13.         //driver端添加app的持有 比如 在waitingApps 中添加 这个 app 为后续调度做准备
  14.         registerApplication(app)
  15.         logInfo("Registered app " + description.name + " with ID " + app.id)
  16.         //持久化这个app
  17.         persistenceEngine.addApplication(app)
  18.         //给driver返回一个已经注册的响应
  19.         driver.send(RegisteredApplication(app.id, self))
  20.         //调度，开始在 Worker 上 分配 executors
  21.         schedule()
  22.       }
  23.   }
  25.   //创建一个app
  26.   private def createApplication(desc: ApplicationDescription, driver: RpcEndpointRef):
  27.       ApplicationInfo = {
  28.     val now = System.currentTimeMillis()
  29.     val date = new Date(now)
  30.     val appId = newApplicationId(date)
  31.     //且会自己执行 init()
  32.     new ApplicationInfo(now, appId, desc, date, driver, defaultCores)
  33.   }
  35.  private def schedule(): Unit = {
  36.     if (state != RecoveryState.ALIVE) {
  37.       return
  38.     }
  39.     val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
  40.     val numWorkersAlive = shuffledAliveWorkers.size
  41.     var curPos = 0
  42.     //启动等待的所有 driver 目前 driver 已经启动了，跳过这一步
  43.     for (driver <- waitingDrivers.toList) { 
  44.       ...
  45.     }
  46.     //在Workers上启动Executors
  47.     startExecutorsOnWorkers()
  48.   }
  50.   //调度并在 workers 上 启动 Executors
  51.  private def startExecutorsOnWorkers(): Unit = {
  52.     //现在这是一个非常简单的FIFO调度器。依次对等待的 app 进行调度 
  53.     for (app <- waitingApps) {
  54.       val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1)
  55.       // 如果剩余的内核小于coresPerExecutor，则不会分配剩余的内核
  56.       // 简单理解就是 app剩下的核 要满足最少是一个 executor 所需的核数 ，也就是以 executor 所需的核数为单位 进行分配 executor 最少的 核数为 1
  57.       if (app.coresLeft >= coresPerExecutor) {
  58.         //过滤掉哪些 没有足够资源 的 worker 并按照剩余的核数倒序排序  来依次启动 executors
  59.         val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
  60.           .filter(canLaunchExecutor(_, app.desc))
  61.           .sortBy(_.coresFree).reverse    
  62.         val appMayHang = waitingApps.length == 1 &&
  63.           waitingApps.head.executors.isEmpty && usableWorkers.isEmpty
  64.         if (appMayHang) {
  65.           logWarning(s"App ${app.id} requires more resource than any of Workers could have.")
  66.         }
  67.         //真正去计算资源分配  返回要给 数组 里面有分配好的核数
  68.         val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps)
  70.         //现在我们已经决定了在每个worker上分配多少个内核，让我们分配它们
  71.         for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
  72.           allocateWorkerResourceToExecutors(
  73.             app, assignedCores(pos), app.desc.coresPerExecutor, usableWorkers(pos))
  74.         }
  75.       }
  76.     }
  77.   }
  79.   //调度 executors 去 workers 上启动。返回一个数组，其中包含分配给每个worker的核心数。
  80.   //有两种启动executor的模式。
  81.   //    第一种方法试图将应用程序的executor分散到尽可能多的worker进程上，默认设置，更适合数据局部性
  82.   //    第二种方法则相反（即在尽可能少的worker进程中启动它们）。
  83.   //分配给每个executor的核心数量是可配置的。当显式设置此选项时，如果worker进程有足够的内核和内存，
  84.   //则可以在同一worker进程上启动来自同一应用程序的多个executor。
  85.   //否则，默认情况下，每个executor都会抓取worker上可用的所有内核，
  86.   //  在这种情况下，在一次单独的计划迭代中，每个应用程序只能在每个worker上启动一个executor。
  87.   //请注意，当未设置“spark.executor.cores”时，
  88.   //  我们仍然可以在同一个worker上从同一个应用程序启动多个executor。
  89.   //  假设appA和appB都有一个executor在worker1上运行，并且appA.coresLef>0，则appB完成并释放worker1上的所有内核，
  90.   //  因此对于下一个计划迭代，appA启动一个新的executor，抓取worker1上所有空闲的内核，因此我们从运行在worker1的appA中获得多个executor。
  91.   private def scheduleExecutorsOnWorkers(
  92.       app: ApplicationInfo,
  93.       usableWorkers: Array[WorkerInfo],
  94.       spreadOutApps: Boolean): Array[Int] = {
  95.     //每个Executor所需的核数
  96.     val coresPerExecutor = app.desc.coresPerExecutor
  97.     //每个Executor所需的最小核数
  98.     val minCoresPerExecutor = coresPerExecutor.getOrElse(1)
  99.     //每个worker一个Executor
  100.     val oneExecutorPerWorker = coresPerExecutor.isEmpty
  101.     //每个Executor所需的内存
  102.     val memoryPerExecutor = app.desc.memoryPerExecutorMB
  103.     //每个Executor所需的资源
  104.     val resourceReqsPerExecutor = app.desc.resourceReqsPerExecutor
  105.     //可用的worker数量
  106.     val numUsable = usableWorkers.length
  107.     val assignedCores = new Array[Int](numUsable) // 去每个worker上需要申请的核数
  108.     val assignedExecutors = new Array[Int](numUsable) // 去每个worker上需要申请的executor数量
  109.     var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
  111.     /**返回指定的worker是否可以为此app启动Executor */
  112.     def canLaunchExecutorForApp(pos: Int): Boolean = {
  113.       val keepScheduling = coresToAssign >= minCoresPerExecutor
  114.       val enoughCores = usableWorkers(pos).coresFree - assignedCores(pos) >= minCoresPerExecutor
  115.       val assignedExecutorNum = assignedExecutors(pos)
  117.       //如果我们允许每个worker有多个executor，那么我们总是可以启动新的executor。
  118.       //否则，如果这个worker上已经有一个executor，只需给它更多的内核。
  119.       val launchingNewExecutor = !oneExecutorPerWorker || assignedExecutorNum == 0
  120.       if (launchingNewExecutor) {
  121.         val assignedMemory = assignedExecutorNum * memoryPerExecutor
  122.         val enoughMemory = usableWorkers(pos).memoryFree - assignedMemory >= memoryPerExecutor
  123.         val assignedResources = resourceReqsPerExecutor.map {
  124.           req => req.resourceName -> req.amount * assignedExecutorNum
  125.         }.toMap
  126.         val resourcesFree = usableWorkers(pos).resourcesAmountFree.map {
  127.           case (rName, free) => rName -> (free - assignedResources.getOrElse(rName, 0))
  128.         }
  129.         val enoughResources = ResourceUtils.resourcesMeetRequirements(
  130.           resourcesFree, resourceReqsPerExecutor)
  131.         val underLimit = assignedExecutors.sum + app.executors.size < app.executorLimit
  132.         keepScheduling && enoughCores && enoughMemory && enoughResources && underLimit
  133.       } else {
  134.         //我们正在为现有的执行器添加内核，因此无需检查内存和执行器限制
  135.         keepScheduling && enoughCores
  136.       }
  137.     }
  139.     //过滤可以启动executor 的worker
  140.     var freeWorkers = (0 until numUsable).filter(canLaunchExecutorForApp)
  141.     //对可以启动executor 的worker循环进行 executors 的分配
  142.     while (freeWorkers.nonEmpty) {
  143.       freeWorkers.foreach { pos =>
  144.         var keepScheduling = true
  145.         while (keepScheduling && canLaunchExecutorForApp(pos)) {
  146.           coresToAssign -= minCoresPerExecutor
  147.           assignedCores(pos) += minCoresPerExecutor
  149.           if (oneExecutorPerWorker) {
  150.             assignedExecutors(pos) = 1
  151.           } else {
  152.             assignedExecutors(pos) += 1
  153.           }
  155.           //分散app意味着将其executor分散到尽可能多的worker中。如果我们不分散，
  156.           //那么我们应该继续在这个worker上调度executor，直到我们使用了它的所有资源。
  157.           //否则，请转到下一个worker。 默认 keepScheduling  = true 
  158.           if (spreadOutApps) {
  159.             keepScheduling = false
  160.           }
  161.         }
  162.       }
  163.       freeWorkers = freeWorkers.filter(canLaunchExecutorForApp)
  164.     }
  165.     assignedCores
  166.   }
  168.   //将Worke的资源分配给一个或多个Executor。
  169.   private def allocateWorkerResourceToExecutors(
  170.       app: ApplicationInfo,
  171.       assignedCores: Int,
  172.       coresPerExecutor: Option[Int],
  173.       worker: WorkerInfo): Unit = {
  174.     //如果指定了每个Executor的Cores，我们将分配给此worker的核心平均分配给Executor 没有余数。
  175.     //否则，我们将启动一个Executor，获取此worker上所有分配的Cores。
  176.     val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)
  177.     val coresToAssign = coresPerExecutor.getOrElse(assignedCores)
  178.     //一个一个executor去worker上启动
  179.     for (i <- 1 to numExecutors) {
  180.       val allocated = worker.acquireResources(app.desc.resourceReqsPerExecutor)
  181.       //这里面会创建一个 ExecutorDesc 作为 Executor 启动的描述 像之前的app、driver都有这个描述
  182.       val exec = app.addExecutor(worker, coresToAssign, allocated)
  183.       //启动Executor
  184.       launchExecutor(worker, exec)
  185.       app.state = ApplicationState.RUNNING
  186.     }
  187.   }
  189.   private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc): Unit = {
  190.     //在那个 worker 上启动 executor 
  191.     logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
  192.     worker.addExecutor(exec)
  193.     //Master endpoint 向 worker endpoint 发送 LaunchExecutor 消息
  194.     worker.endpoint.send(LaunchExecutor(masterUrl, exec.application.id, exec.id,
  195.       exec.application.desc, exec.cores, exec.memory, exec.resources))
  196.     //Master endpoint 向 driverendpoint 发送 ExecutorAdded消息
  197.     exec.application.driver.send(
  198.       ExecutorAdded(exec.id, worker.id, worker.hostPort, exec.cores, exec.memory))
  199.   }
  201. }

九、Worker

  1. private[deploy] class Worker(......
  2.   extends ThreadSafeRpcEndpoint with Logging {
  3.   override def receive: PartialFunction[Any, Unit] = synchronized {
  4.    case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_, resources_) =>
  5.       if (masterUrl != activeMasterUrl) {
  6.         logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor.")
  7.       } else if (decommissioned) {
  8.         logWarning("Asked to launch an executor while decommissioned. Not launching executor.")
  9.       } else {
  10.         try {
  11.           logInfo("Asked to launch executor %s/%d for %s".format(appId, execId, appDesc.name))
  13.           // 创建 executor 的 本地工作目录
  14.           val executorDir = new File(workDir, appId + "/" + execId)
  15.           if (!executorDir.mkdirs()) {
  16.             throw new IOException("Failed to create directory " + executorDir)
  17.           }
  19.           // 为执行者创建本地目录。这些通过SPARK_EXECUTOR_DIRS环境变量传递给executor，并在应用程序完成时由Worker删除。
  20.           val appLocalDirs = appDirectories.getOrElse(appId, {
  21.             val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
  22.             val dirs = localRootDirs.flatMap { dir =>
  23.               try {
  24.                 val appDir = Utils.createDirectory(dir, namePrefix = "executor")
  25.                 Utils.chmod700(appDir)
  26.                 Some(appDir.getAbsolutePath())
  27.               } catch {
  28.                 ...
  29.               }
  30.             }.toSeq
  31.             dirs
  32.           })
  33.           appDirectories(appId) = appLocalDirs
  34.           //管理一个executor流程的执行。这目前仅在standalone模式下使用。
  35.           val manager = new ExecutorRunner(
  36.             appId,
  37.             execId,
  38.             appDesc.copy(command = Worker.maybeUpdateSSLSettings(appDesc.command, conf)),
  39.             cores_,
  40.             memory_,
  41.             self,
  42.             workerId,
  43.             webUi.scheme,
  44.             host,
  45.             webUi.boundPort,
  46.             publicAddress,
  47.             sparkHome,
  48.             executorDir,
  49.             workerUri,
  50.             conf,
  51.             appLocalDirs,
  52.             ExecutorState.LAUNCHING,
  53.             resources_)
  54.           executors(appId + "/" + execId) = manager
  55.           manager.start()
  56.           coresUsed += cores_
  57.           memoryUsed += memory_
  58.           addResourcesUsed(resources_)
  59.         } catch {
  60.           ......
  61.         }
  62.       }
  63.   }
  65. }

十、ExecutorRunner

  1. private[deploy] class ExecutorRunner(...){
  3.   private[worker] def start(): Unit = {
  4.     //准备一个线程启动 executor 
  5.     workerThread = new Thread("ExecutorRunner for " + fullId) {
  6.       override def run(): Unit = { fetchAndRunExecutor() }
  7.     }
  8.     //线程启动 fetchAndRunExecutor() 执行 ：下载并运行应用程序描述中描述的executor 
  9.     //第6步中封装的启动类为 org.apache.spark.executor.CoarseGrainedExecutorBacken
  10.     //下面我们看看 executor 中做了什么
  11.     workerThread.start()
  12.     ......
  13.   }
  15. }

十一、CoarseGrainedExecutorBacken（executor进程主类）

  1. private[spark] object CoarseGrainedExecutorBackend extends Logging {
  3.   def main(args: Array[String]): Unit = {
  4.     val createFn: (RpcEnv, Arguments, SparkEnv, ResourceProfile) =>
  5.       CoarseGrainedExecutorBackend = { case (rpcEnv, arguments, env, resourceProfile) =>
  6.       new CoarseGrainedExecutorBackend(rpcEnv, arguments.driverUrl, arguments.executorId,
  7.         arguments.bindAddress, arguments.hostname, arguments.cores, arguments.userClassPath.toSeq,
  8.         env, arguments.resourcesFileOpt, resourceProfile)
  9.     }
  10.     //解析参数并运行
  11.     run(parseArguments(args, this.getClass.getCanonicalName.stripSuffix("$")), createFn)
  12.     System.exit(0)
  13.   }
  15.  def run(
  16.       arguments: Arguments,
  17.       backendCreateFn: (RpcEnv, Arguments, SparkEnv, ResourceProfile) =>
  18.         CoarseGrainedExecutorBackend): Unit = {
  20.       //......
  22.       // 创建 RpcEnv 获取 driver端的 Spark properties.
  23.       val executorConf = new SparkConf
  24.       val fetcher = RpcEnv.create(
  25.         "driverPropsFetcher",
  26.         arguments.bindAddress,
  27.         arguments.hostname,
  28.         -1,
  29.         executorConf,
  30.         new SecurityManager(executorConf),
  31.         numUsableCores = 0,
  32.         clientMode = true)
  34.       //尝试3次获取 driver Endpoint 引用
  35.       var driver: RpcEndpointRef = null
  36.       val nTries = 3
  37.       for (i <- 0 until nTries if driver == null) {
  38.         try {
  39.           driver = fetcher.setupEndpointRefByURI(arguments.driverUrl)
  40.         } catch {
  41.           case e: Throwable => if (i == nTries - 1) {
  42.             throw e
  43.           }
  44.         }
  45.       }
  47.       // 向driver发送 RetrieveSparkAppConfig 消息
  48.       val cfg = driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(arguments.resourceProfileId))
  49.       val props = cfg.sparkProperties ++ Seq[(String, String)](("spark.app.id", arguments.appId))
  50.       fetcher.shutdown()
  52.       //根据从driver端获取的属性创建SparkEnv
  53.       val driverConf = new SparkConf()
  54.       for ((key, value) <- props) {
  55.         // this is required for SSL in standalone mode
  56.         if (SparkConf.isExecutorStartupConf(key)) {
  57.           driverConf.setIfMissing(key, value)
  58.         } else {
  59.           driverConf.set(key, value)
  60.         }
  61.       }
  63.       cfg.hadoopDelegationCreds.foreach { tokens =>
  64.         SparkHadoopUtil.get.addDelegationTokens(tokens, driverConf)
  65.       }
  67.       driverConf.set(EXECUTOR_ID, arguments.executorId)
  68.       //为executor创建SparkEnv
  69.       val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId, arguments.bindAddress,
  70.         arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal = false)
  71.       // 在BlockStoreClient中设置应用程序尝试ID（如果可用）
  72.       val appAttemptId = env.conf.get(APP_ATTEMPT_ID)
  73.       appAttemptId.foreach(attemptId =>
  74.         env.blockManager.blockStoreClient.setAppAttemptId(attemptId)
  75.       )
  76.       //创建CoarseGrainedExecutorBackend
  77.       val backend = backendCreateFn(env.rpcEnv, arguments, env, cfg.resourceProfile)
  78.       //并将CoarseGrainedExecutorBackend设置为 Executor Endpoint 其上的 onStart() 方法执行
  79.       env.rpcEnv.setupEndpoint("Executor", backend)
  80.       arguments.workerUrl.foreach { url =>
  81.         env.rpcEnv.setupEndpoint("WorkerWatcher",
  82.           new WorkerWatcher(env.rpcEnv, url, isChildProcessStopping = backend.stopping))
  83.       }
  84.       env.rpcEnv.awaitTermination()
  85.     }
  86.   }
  88. }

十二、CoarseGrainedExecutorBackend（executor中的Rpc端点）

  1. private[spark] class CoarseGrainedExecutorBackend(...)
  2.   extends IsolatedRpcEndpoint with ExecutorBackend with Logging {
  3.   //当driver成功接受注册请求时，内部用于启动executor的消息。
  4.   case object RegisteredExecutor
  6.   override def onStart(): Unit = {
  7.     //......
  8.     logInfo("Connecting to driver: " + driverUrl)
  9.     try {
  10.       //提供了一个实用程序，用于将Spark JVM内的SparkConf（例如，执行器、驱动程序或独立的shuffle服务）转换为TransportConf，其中包含有关我们环境的详细信息，如分配给此JVM的内核数量。
  11.       val shuffleClientTransportConf = SparkTransportConf.fromSparkConf(env.conf, "shuffle") 
  12.       //判断Netty是否可以直接使用 off-heap 内存  且 操作系统能分配的最大 off-heap < 200M 抛出异常
  13.       //涉及配置  
  14.       //    spark.network.sharedByteBufAllocators.enabled 默认true 是否在不同Netty通道之间共享池化ByteBuf分配器的标志。如果启用，则只创建两个池化ByteBuf分配器：一个允许缓存（用于传输服务器），另一个不允许缓存（对于传输客户端）。禁用后，将为每个传输服务器和客户端创建一个新的分配器。
  15.       //    spark.io.preferDirectBufs  默认true 共享ByteBuf分配器将首选堆外字节缓冲区
  16.       //    spark.network.io.preferDirectBufs 默认true 在Netty中分配堆外字节缓冲区
  17.       if (NettyUtils.preferDirectBufs(shuffleClientTransportConf) &&
  18.           PlatformDependent.maxDirectMemory() < env.conf.get(MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM)) {
  19.         throw new SparkException(s"Netty direct memory should at least be bigger than " +
  20.           s"'${MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM.key}', but got " +
  21.           s"${PlatformDependent.maxDirectMemory()} bytes < " +
  22.           s"${env.conf.get(MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM)}")
  23.       }
  25.       _resources = parseOrFindResources(resourcesFileOpt)
  26.     } catch {
  27.       case NonFatal(e) =>
  28.         exitExecutor(1, "Unable to create executor due to " + e.getMessage, e)
  29.     }
  30.     rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>
  31.       //这是一个非常快速的操作，因此我们可以使用“ThreadUtils.sameThread”
  32.       driver = Some(ref)
  33.       //向driver 发送 RegisterExecutor 消息 ，我们看第五步中driver的处理
  34.       ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores, extractLogUrls,
  35.         extractAttributes, _resources, resourceProfile.id))
  36.     }(ThreadUtils.sameThread).onComplete {
  37.       case Success(_) =>
  38.         self.send(RegisteredExecutor)
  39.       case Failure(e) =>
  40.         exitExecutor(1, s"Cannot register with driver: $driverUrl", e, notifyDriver = false)
  41.     }(ThreadUtils.sameThread)
  42.   }
  44.   override def receive: PartialFunction[Any, Unit] = {
  45.     //driver收到注册 executor 并返回 true 执行该段逻辑
  46.     case RegisteredExecutor =>
  47.       logInfo("Successfully registered with driver")
  48.       try {
  49.         //创建一个 Executor
  50.         executor = new Executor(executorId, hostname, env, userClassPath, isLocal = false,
  51.           resources = _resources)
  52.         //再向driver发送 LaunchedExecutor 消息，可以看 第五步 driver端的处理
  53.         driver.get.send(LaunchedExecutor(executorId))
  54.       } catch {
  55.         case NonFatal(e) =>
  56.           exitExecutor(1, "Unable to create executor due to " + e.getMessage, e)
  57.       }
  58.   }
  60. }

十三、Executor

  1. //Spark执行器，由线程池支持运行任务。
  2. //这可以与Mesos、YARN、kubernetes和独立调度器一起使用。内部RPC接口用于与driver通信，Mesos细粒度模式除外。
  3. private[spark] class Executor(...){
  5.   // 维护正在运行的任务列表
  6.   private val runningTasks = new ConcurrentHashMap[Long, TaskRunner]
  8.   //当Executor无法向driver程序发送心跳超过“HEARTBEAT_MAX_FAILURES”次数时，它应该自行终止。默认值为60。例如，如果最大失败次数为60次，心跳间隔为10秒，则它将尝试发送长达600秒（10分钟）的心跳。
  9.   private val HEARTBEAT_MAX_FAILURES = conf.get(EXECUTOR_HEARTBEAT_MAX_FAILURES)
  10.   //发送心跳的间隔（毫秒） spark.executor.heartbeatInterval 默认 10s
  11.   private val HEARTBEAT_INTERVAL_MS = conf.get(EXECUTOR_HEARTBEAT_INTERVAL)
  12.   //heartbeat 任务
  13.   private val heartbeater = new Heartbeater(
  14.     () => Executor.this.reportHeartBeat(),
  15.     "executor-heartbeater",
  16.     HEARTBEAT_INTERVAL_MS)
  18.   //启动工作线程池
  19.   private val threadPool = {
  20.     val threadFactory = new ThreadFactoryBuilder()
  21.       .setDaemon(true)
  22.       .setNameFormat("Executor task launch worker-%d")
  23.       .setThreadFactory((r: Runnable) => new UninterruptibleThread(r, "unused"))
  24.       .build()
  25.     Executors.newCachedThreadPool(threadFactory).asInstanceOf[ThreadPoolExecutor]
  26.   }
  28.   //启动任务
  29.   def launchTask(context: ExecutorBackend, taskDescription: TaskDescription): Unit = {
  30.     val tr = new TaskRunner(context, taskDescription, plugins)
  31.     runningTasks.put(taskDescription.taskId, tr)
  32.     threadPool.execute(tr)
  33.     if (decommissioned) {
  34.       log.error(s"Launching a task while in decommissioned state.")
  35.     }
  36.   }
  38. }

十四、总结

1、代码中根据SparkConf构建SparkContext

2、创建任务调度器并启用

3、StandaloneSchedulerBackend 和 CoarseGrainedSchedulerBackend 的 start() 启动

4、DriverEndpoint 创建 等待其他Endpoint发送消息 （比如Master 和 Executur Endpoint）

5、构建Executor的启动参数，主类为CoarseGrainedExecutorBackend

6、创建StandaloneAppClient并启动

7、Driver端创建ClientEndpoint并向Master注册

8、创建app描述信息向Master发送RegisterApplication 消息

9、Master 根据app描述信息开始调度资源，决策在哪些Worker上启动多少个Executor

10、Master端以Executor为单位依次向划分好的Worker发送LaunchExecutor消息，向Driver发送ExecutorAdded消息

11、Worker 创建一个线程启动封装好的Executor进程（主类为CoarseGrainedExecutorBackend）

12、Executor中会创建Executor Endpoint，并向Driver进行注册，如果注册成功会向自己发送RegisteredExecutor消息

13、Executor处理给自己发的RegisteredExecutor消息，其中会创建一个Executor对象并向Driver发送LaunchedExecutor消息

14、Executor对象由线程池支持运行任务，且并默认每隔10s发送依次心跳给Driver

为了方便理解和记忆，我们也画下流程图，下载放大就会清晰哟

标签： spark 大数据 分布式
本文转载自: https://blog.csdn.net/lu070828/article/details/141355638
版权归原作者 隔着天花板看星星 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Spark-driver和executor启动过程”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？