org.apache.spark.deploy.SparkSubmit
-main
-submit.doSubmit(args)
- SparkSubmitArguments.parseArguments(args)
- - SparkSubmitAction.SUBMIT => submit(appArgs, uninitLog)
- submit(args: SparkSubmitArguments, uninitLog: Boolean)
-runMain(args: SparkSubmitArguments, uninitLog: Boolean)
- new JavaMainApplication(mainClass)
-app.start(childArgs.toArray, sparkConf)
// Following constants are visible for testing.
private[deploy] val YARN_CLUSTER_SUBMIT_CLASS =
"org.apache.spark.deploy.yarn.YarnClusterApplication"
// In yarn-cluster mode, use yarn.Client as a wrapper around the user class
if (isYarnCluster) {
childMainClass = YARN_CLUSTER_SUBMIT_CLASS
//mainClass 是SparkApplication类型,构建SparkApplication
val app: SparkApplication = if (classOf[SparkApplication].isAssignableFrom(mainClass)) {
mainClass.getConstructor().newInstance().asInstanceOf[SparkApplication]
} else {
new JavaMainApplication(mainClass)
}
-YarnClusterApplication.start
//ClientArguments传递这spark submit的提交单数
- new Client(new ClientArguments(args), conf, null).run()
-Client.submitApplication()
private val yarnClient = YarnClient.createYarnClient
//用于与rm通信
protected ApplicationClientProtocol rmClient;
//获取appid
protected ApplicationClientProtocol rmClient;
//穿件容器上下文环境
val containerContext = createContainerLaunchContext(newAppResponse)
containerContext 上下文包含 运行ApplicationMaster的的classname 等启动参数
val amClass =
if (isClusterMode) {
Utils.classForName("org.apache.spark.deploy.yarn.ApplicationMaster").getName
} else {
Utils.classForName("org.apache.spark.deploy.yarn.ExecutorLauncher").getName
}
//创建提交app的上下文环境
val appContext = createApplicationSubmissionContext(newApp, containerContext)
-yarnClient.submitApplication(appContext)
-rmClient.submitApplication(request);
搜索”org.apache.spark.deploy.yarn.ApplicationMaster“
-org.apache.spark.deploy.yarn.ApplicationMaster.main
-master = new ApplicationMaster(amArgs, sparkConf, yarnConf)
ApplicationMaster成员
private val client = new YarnRMClient()
//用于am和rm通信
-amClient = AMRMClient.createAMRMClient()
-master = new ApplicationMaster(amArgs, sparkConf, yarnConf)
-master.run()
如果是集群模式,运行Driver
if (isClusterMode) {
runDriver()
} else {
runExecutorLauncher()
}
private def runDriver(): Unit = {
addAmIpFilter(None, System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV))
userClassThread = startUserApplication()
... ...
try {
//等待startUserApplication的中driver线程将用户代码sparkContext的创建完成,否则一直阻塞在这里
val sc = ThreadUtils.awaitResult(sparkContextPromise.future,
Duration(totalWaitTime, TimeUnit.MILLISECONDS))
//sc不为空
if (sc != null) {
val rpcEnv = sc.env.rpcEnv
val userConf = sc.getConf
val host = userConf.get(DRIVER_HOST_ADDRESS)
val port = userConf.get(DRIVER_PORT)
//注册AM,申请西苑
registerAM(host, port, userConf, sc.ui.map(_.webUrl), appAttemptId)
val driverRef = rpcEnv.setupEndpointRef(
RpcAddress(host, port),
YarnSchedulerBackend.ENDPOINT_NAME)
//创建分配器,返回资源可用列表
createAllocator(driverRef, userConf, rpcEnv, appAttemptId, distCacheConf)
} else {
// Sanity check; should never happen in normal operation, since sc should only be null
// if the user app did not create a SparkContext.
throw new IllegalStateException("User did not initialize spark context!")
}
//当资源准备就绪,调用resumeDriver方法,改变状态,让driver线程继续执行(用户代码逻辑)
resumeDriver()
userClassThread执行完之后,rundriver方法再继续执行
userClassThread.join()
} catch {
... ...
} finally {
resumeDriver()
}
}
userClassThread = startUserApplication()
private def startUserApplication(): Thread = {
logInfo("Starting the user application in a separate Thread")
... ...
val mainMethod = userClassLoader.loadClass(args.userClass)
.getMethod("main", classOf[Array[String]])
val userThread = new Thread {
override def run(): Unit = {
try {
if (!Modifier.isStatic(mainMethod.getModifiers)) {
logError(s"Could not find static main method in object ${args.userClass}")
finish(FinalApplicationStatus.FAILED, ApplicationMaster.EXIT_EXCEPTION_USER_CLASS)
} else {
//静态方法,也就是用户的用户编写Job Object 的main方法,调用用户代码
mainMethod.invoke(null, userArgs.toArray)
finish(FinalApplicationStatus.SUCCEEDED, ApplicationMaster.EXIT_SUCCESS)
logDebug("Done running user class")
}
} catch {
......
} ... ...
}
}
userThread.setContextClassLoader(userClassLoader)
//设置drive线程
userThread.setName("Driver")
//设置drive线程启动
userThread.start()
userThread
}
创建SparkContext
val sc = new SparkContext(sparConf)
SparkContext进行sc初始化的时候,有一段代码如下
// Post init
_taskScheduler.postStartHook()
private[spark] class YarnClusterScheduler(sc: SparkContext) extends YarnScheduler(sc) {
...
override def postStartHook(): Unit = {
//sparkContextInitialized干两个事儿
//SparkContext 初始化后,唤醒runDriver 方法继续执行
//暂停driver线程(user thread),为了让在runDriver函数中进行初始化
ApplicationMaster.sparkContextInitialized(sc)
super.postStartHook()
logInfo("YarnClusterScheduler.postStartHook done")
}
}
private def sparkContextInitialized(sc: SparkContext) = {
sparkContextPromise.synchronized {
//SparkContext 初始化后,唤醒runDriver 方法继续执行
// Notify runDriver function that SparkContext is available
sparkContextPromise.success(sc)
//暂停driver线程(user thread),为了让在runDriver函数中进行初始化
// Pause the user class thread in order to make proper initialization in runDriver function.
sparkContextPromise.wait()
}
}
-TaskSchedulerImpl.postStartHook
TaskSchedulerImpl.waitBackendReady
private def waitBackendReady(): Unit = {
if (backend.isReady) {
return
}
//循环等待知道资源就绪,此时用户代码不会往下执行
//那什么时候driver线程会继续执行?
//当rundrive方法调用resumeDriver,改变backend 状态,代表资源就绪
while (!backend.isReady) {
// Might take a while for backend to be ready if it is waiting on resources.
if (sc.stopped.get) {
// For example: the master removes the application for some reason
throw new IllegalStateException("Spark context stopped while waiting for backend")
}
synchronized {
this.wait(100)
}
}
}
runDriver方法的registerAM
private val client = new YarnRMClient()
client.register(host, port, yarnConf, _sparkConf, uiAddress, historyAddress)
amClient = AMRMClient.createAMRMClient()
通过AMRMClient向RM注册
amClient.registerApplicationMaster(driverHost, driverPort, trackingUrl)
private def createAllocator(
driverRef: RpcEndpointRef,
_sparkConf: SparkConf,
rpcEnv: RpcEnv,
appAttemptId: ApplicationAttemptId,
distCacheConf: SparkConf): Unit = {
... ...
val appId = appAttemptId.getApplicationId().toString()
val driverUrl = RpcEndpointAddress(driverRef.address.host, driverRef.address.port,
CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
val localResources = prepareLocalResources(distCacheConf)
... ...
//创建分配器
allocator = client.createAllocator(
yarnConf,
_sparkConf,
appAttemptId,
driverUrl,
driverRef,
securityMgr,
localResources)
... ...
//获取可用资源列表
allocator.allocateResources()
val ms = MetricsSystem.createMetricsSystem(MetricsSystemInstances.APPLICATION_MASTER,
sparkConf, securityMgr)
val prefix = _sparkConf.get(YARN_METRICS_NAMESPACE).getOrElse(appId)
ms.registerSource(new ApplicationMasterSource(prefix, allocator))
// do not register static sources in this case as per SPARK-25277
ms.start(false)
metricsSystem = Some(ms)
reporterThread = launchReporterThread()
}
//处理可用于分配的容器
handleAllocatedContainers(allocatedContainers.asScala.toSeq)
def handleAllocatedContainers(allocatedContainers: Seq[Container]): Unit = {
val containersToUse = new ArrayBuffer[Container](allocatedContainers.size)
// Match incoming requests by host
val remainingAfterHostMatches = new ArrayBuffer[Container]
//可分配容器 分类整理按主机名和机架(首选位置的应用)
for (allocatedContainer <- allocatedContainers) {
matchContainerToRequest(allocatedContainer, allocatedContainer.getNodeId.getHost,
containersToUse, remainingAfterHostMatches)
}
... ...
val remainingAfterRackMatches = new ArrayBuffer[Container]
if (remainingAfterHostMatches.nonEmpty) {
var exception: Option[Throwable] = None
val thread = new Thread("spark-rack-resolver") {
override def run(): Unit = {
try {
for (allocatedContainer <- remainingAfterHostMatches) {
val rack = resolver.resolve(allocatedContainer.getNodeId.getHost)
matchContainerToRequest(allocatedContainer, rack, containersToUse,
remainingAfterRackMatches)
}
} catch {
case e: Throwable =>
... ...
}
//运行已分配容器进行
runAllocatedContainers(containersToUse)
logInfo("Received %d containers from YARN, launching executors on %d of them."
.format(allocatedContainers.size, containersToUse.size))
}
runAllocatedContainers 启动容器
private def runAllocatedContainers(containersToUse: ArrayBuffer[Container]): Unit = synchronized {
for (container <- containersToUse) {
val rpId = getResourceProfileIdFromPriority(container.getPriority)
executorIdCounter += 1
val executorHostname = container.getNodeId.getHost
val containerId = container.getId
val executorId = executorIdCounter.toString
val yarnResourceForRpId = rpIdToYarnResource.get(rpId)
assert(container.getResource.getMemory >= yarnResourceForRpId.getMemory)
logInfo(s"Launching container $containerId on host $executorHostname " +
s"for executor with ID $executorId for ResourceProfile Id $rpId")
... ...
TargetNum容器大于Running的容器数,说明还需要启动容器
if (rpRunningExecs < getOrUpdateTargetNumExecutorsForRPId(rpId)) {
getOrUpdateNumExecutorsStartingForRPId(rpId).incrementAndGet()
if (launchContainers) {
启动线程池,启动容器
launcherPool.execute(() => {
try {
new ExecutorRunnable(
Some(container),
conf,
sparkConf,
driverUrl,
executorId,
executorHostname,
containerMem,
containerCores,
appAttemptId.getApplicationId.toString,
securityMgr,
localResources,
rp.id
).run()
updateInternalState()
} catch {
... ...
}
})
} else {
// For test only
updateInternalState()
}
... ...
}
}
ExecutorRunnable.run
//
nmClient = NMClient.createNMClient()
nmClient.init(conf)
nmClient.start()
//通过nmClient 通知指定nm启动contanter
startContainer()
startContainer
def startContainer(): java.util.Map[String, ByteBuffer] = {
... ...
//prepareCommand 是准备启动容器进程的脚本
//启动org.apache.spark.executor.YarnCoarseGrainedExecutorBackend 进程(excutor 的通信后台)
val commands = prepareCommand()
ctx.setCommands(commands.asJava)
ctx.setApplicationACLs(
YarnSparkHadoopUtil.getApplicationAclsForYarn(securityMgr).asJava)
... ....
// Send the start request to the ContainerManager
try {
//启动container,携带启动容器的上下文ctx
nmClient.startContainer(container.get, ctx)
} catch {
... ...
}
}
org.apache.spark.executor.YarnCoarseGrainedExecutorBackend
-CoarseGrainedExecutorBackend.run
def run(
arguments: Arguments,
backendCreateFn: (RpcEnv, Arguments, SparkEnv, ResourceProfile) =>
CoarseGrainedExecutorBackend): Unit = {
... ...
//可以找到driver 与driver通信
val fetcher = RpcEnv.create(
"driverPropsFetcher",
arguments.bindAddress,
arguments.hostname,
-1,
executorConf,
new SecurityManager(executorConf),
numUsableCores = 0,
clientMode = true)
var driver: RpcEndpointRef = null
val nTries = 3
for (i <- 0 until nTries if driver == null) {
try {
driver = fetcher.setupEndpointRefByURI(arguments.driverUrl)
} catch {
case e: Throwable => if (i == nTries - 1) {
throw e
}
}
}
//创建Executor的evn环境
val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId, arguments.bindAddress,
arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal = false)
//设置rpcEnv环境的通信终端
backendCreateFn 其实就是YarnCoarseGrainedExecutorBackend
env.rpcEnv.setupEndpoint("Executor",
backendCreateFn(env.rpcEnv, arguments, env, cfg.resourceProfile))
arguments.workerUrl.foreach { url =>
env.rpcEnv.setupEndpoint("WorkerWatcher", new WorkerWatcher(env.rpcEnv, url))
}
env.rpcEnv.awaitTermination()
}
}
//消息循环器
var messageLoop: MessageLoop = null
try {
messageLoop = endpoint match {
case e: IsolatedRpcEndpoint =>
//匹配消息循环器类型
new DedicatedMessageLoop(name, e, this)
case _ =>
sharedLoop.register(name, endpoint)
sharedLoop
}
endpoints.put(name, messageLoop)
} catch {
case NonFatal(e) =>
endpointRefs.remove(endpoint)
throw e
}
-DedicatedMessageLoop
//收件箱
private val inbox = new Inbox(name, endpoint)
-Inbox
protected val messages = new java.util.LinkedListInboxMessage
// OnStart should be the first message to process
//放入一个OnStart消息
inbox.synchronized {
messages.add(OnStart)
}
//RpcEndpoint的生命周期
* {@code constructor -> onStart -> receive* -> onStop}
private[spark] trait RpcEndpoint {
CoarseGrainedExecutorBackend.onStart
override def onStart(): Unit = {
... ...
logInfo("Connecting to driver: " + driverUrl)
try {
_resources = parseOrFindResources(resourcesFileOpt)
} catch {
case NonFatal(e) =>
exitExecutor(1, "Unable to create executor due to " + e.getMessage, e)
}
rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>
// This is a very fast action so we can use "ThreadUtils.sameThread"
//获取driver
driver = Some(ref)
//像driver发送RegisterExecutor消息
ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores, extractLogUrls,
extractAttributes, _resources, resourceProfile.id))
}(ThreadUtils.sameThread).onComplete {
case Success(_) =>
self.send(RegisteredExecutor)
case Failure(e) =>
exitExecutor(1, s"Cannot register with driver: $driverUrl", e, notifyDriver = false)
}(ThreadUtils.sameThread)
}
driver是一个线程,所有是SparkContext接收消息
_schedulerBackend是driver通信后台
private var _schedulerBackend: SchedulerBackend = _
//集群模式的SchedulerBackend
CoarseGrainedSchedulerBackend
//消息回复
CoarseGrainedSchedulerBackend.receiveAndReply
匹配RegisterExecutor消息
case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls,
attributes, resources, resourceProfileId) =>
总的核数 注册数量增加
totalCoreCount.addAndGet(cores)
totalRegisteredExecutors.addAndGet(1)
//最后回复一个true表示注册成功
context.reply(true)
CoarseGrainedExecutorBackend.收到消息
case Success(_) =>
//给自己发送一条消息,表示注册完毕
self.send(RegisteredExecutor)
CoarseGrainedExecutorBackend.收到给自己发送的RegisteredExecutor消息
override def receive: PartialFunction[Any, Unit] = {
case RegisteredExecutor =>
logInfo("Successfully registered with driver")
try {
//创建Executor计算对象,区别CoarseGrainedExecutorBackend（通信对象）
executor = new (executorId, hostname, env, userClassPath, isLocal = false,
resources = _resources)
//给driver发送LaunchedExecutor 消息
driver.get.send(LaunchedExecutor(executorId))
driver端的CoarseGrainedSchedulerBackend receive到
LaunchedExecutor消息
case LaunchedExecutor(executorId) =>
//增加核数
executorDataMap.get(executorId).foreach { data =>
data.freeCores = data.totalCores
}
//做一些操作 //tode
makeOffers(executorId)
case e =>
logError(s"Received unexpected message. ${e}")
