图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Spark › 代码分析› Spark源码系列（一）spark-submit提交作业过程
分享

Spark源码系列（一）spark-submit提交作业过程

sunshine_junge 2014-7-7 20:39:49 发表于 代码分析 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 114122
本帖最后由 pig2 于 2014-9-14 12:41 编辑
问题导读：
1.spark是如何提交作业的？
2.Akka框架是如何实现的？
3.如何实现调度的？










前言

折腾了很久，终于开始学习Spark的源码了，第一篇我打算讲一下Spark作业的提交过程。



这个是Spark的App运行图，它通过一个Driver来和集群通信，集群负责作业的分配。今天我要讲的是如何创建这个Driver Program的过程。


作业提交方法以及参数
我们先看一下用Spark Submit提交的方法吧，下面是从官方上面摘抄的内容。


  1. # Run on a Spark standalone cluster

  2. ./bin/spark-submit \

  3.   --class org.apache.spark.examples.SparkPi \

  4.   --master spark://207.184.161.138:7077 \

  5.   --executor-memory 20G \

  6.   --total-executor-cores 100 \

  7.   /path/to/examples.jar \

  8.   1000
复制代码



这个是提交到standalone集群的方式，打开spark-submit这文件，我们会发现它最后是调用了org.apache.spark.deploy.SparkSubmit这个类。

我们直接进去看就行了，main函数就几行代码，太节省了。


  1. def main(args: Array[String]) {

  2.     val appArgs = new SparkSubmitArguments(args)

  3.     val (childArgs, classpath, sysProps, mainClass) = createLaunchEnv(appArgs)

  4.     launch(childArgs, classpath, sysProps, mainClass, appArgs.verbose)

  5. }
复制代码



我们主要看看createLaunchEnv方法就可以了，launch是反射调用mainClass，精华全在createLaunchEnv里面了。

在里面我发现一些有用的信息，可能在官方文档上面都没有的，发出来大家瞅瞅。前面不带--的可以在spark-defaults.conf里面设置，带--的直接在提交的时候指定，具体含义大家一看就懂。

  1. val options = List[OptionAssigner](

  2.       OptionAssigner(args.master, ALL_CLUSTER_MGRS, false, sysProp = "spark.master"),

  3.       OptionAssigner(args.name, ALL_CLUSTER_MGRS, false, sysProp = "spark.app.name"),

  4.       OptionAssigner(args.name, YARN, true, clOption = "--name", sysProp = "spark.app.name"),

  5.       OptionAssigner(args.driverExtraClassPath, STANDALONE | YARN, true,

  6.         sysProp = "spark.driver.extraClassPath"),

  7.       OptionAssigner(args.driverExtraJavaOptions, STANDALONE | YARN, true,

  8.         sysProp = "spark.driver.extraJavaOptions"),

  9.       OptionAssigner(args.driverExtraLibraryPath, STANDALONE | YARN, true,

  10.         sysProp = "spark.driver.extraLibraryPath"),

  11.       OptionAssigner(args.driverMemory, YARN, true, clOption = "--driver-memory"),

  12.       OptionAssigner(args.driverMemory, STANDALONE, true, clOption = "--memory"),

  13.       OptionAssigner(args.driverCores, STANDALONE, true, clOption = "--cores"),

  14.       OptionAssigner(args.queue, YARN, true, clOption = "--queue"),

  15.       OptionAssigner(args.queue, YARN, false, sysProp = "spark.yarn.queue"),

  16.       OptionAssigner(args.numExecutors, YARN, true, clOption = "--num-executors"),

  17.       OptionAssigner(args.numExecutors, YARN, false, sysProp = "spark.executor.instances"),

  18.       OptionAssigner(args.executorMemory, YARN, true, clOption = "--executor-memory"),

  19.       OptionAssigner(args.executorMemory, STANDALONE | MESOS | YARN, false,

  20.         sysProp = "spark.executor.memory"),

  21.       OptionAssigner(args.executorCores, YARN, true, clOption = "--executor-cores"),

  22.       OptionAssigner(args.executorCores, YARN, false, sysProp = "spark.executor.cores"),

  23.       OptionAssigner(args.totalExecutorCores, STANDALONE | MESOS, false,

  24.         sysProp = "spark.cores.max"),

  25.       OptionAssigner(args.files, YARN, false, sysProp = "spark.yarn.dist.files"),

  26.       OptionAssigner(args.files, YARN, true, clOption = "--files"),

  27.       OptionAssigner(args.files, LOCAL | STANDALONE | MESOS, false, sysProp = "spark.files"),

  28.       OptionAssigner(args.files, LOCAL | STANDALONE | MESOS, true, sysProp = "spark.files"),

  29.       OptionAssigner(args.archives, YARN, false, sysProp = "spark.yarn.dist.archives"),

  30.       OptionAssigner(args.archives, YARN, true, clOption = "--archives"),

  31.       OptionAssigner(args.jars, YARN, true, clOption = "--addJars"),

  32.       OptionAssigner(args.jars, ALL_CLUSTER_MGRS, false, sysProp = "spark.jars")

  33.  )
复制代码



Driver程序的部署模式有两种，client和cluster，默认是client。client的话默认就是直接在本地运行了Driver程序了，cluster模式还会兜一圈把作业发到集群上面去运行。

指定部署模式需要用参数--deploy-mode来指定，或者在环境变量当中添加DEPLOY_MODE变量来指定。

下面讲的是cluster的部署方式，兜一圈的这种情况。

yarn模式的话mainClass是org.apache.spark.deploy.yarn.Client，standalone的mainClass是org.apache.spark.deploy.Client。

这次我们讲org.apache.spark.deploy.Client，yarn的话单独找一章出来单独讲，目前超哥还是推荐使用standalone的方式部署spark，具体原因不详，据说是因为资源调度方面的问题。

说个快捷键吧，Ctrl+Shift+N，然后输入Client就能找到这个类，这是IDEA的快捷键，相当好使。

我们直接找到它的main函数，发现了它居然使用了Akka框架，我百度了一下，被它震惊了。


Akka
在main函数里面，主要代码就这么三行。


  1. //创建一个ActorSystem

  2. val (actorSystem, _) = AkkaUtils.createActorSystem("driverClient",Utils.localHostName(),0,

  3. 　　conf, new SecurityManager(conf))

  4. //执行ClientActor的preStart方法和receive方法

  5. actorSystem.actorOf(Props(classOf[ClientActor], driverArgs, conf))

  6. //等待运行结束

  7. actorSystem.awaitTermination()
复制代码



看了这里真的有点儿懵啊，这是啥玩意儿，不懂的朋友们，请点击这里Akka。下面是它官方放出来的例子：


  1. //定义一个case class用来传递参数

  2. case class Greeting(who: String)

  3. //定义Actor，比较重要的一个方法是receive方法，用来接收信息的

  4. class GreetingActor extends Actor with ActorLogging {

  5.    def receive = {

  6.        case Greeting(who) ⇒ log.info("Hello " + who)

  7.    }

  8. }

  9. //创建一个ActorSystem

  10. val system = ActorSystem("MySystem")

  11. //给ActorSystem设置Actor

  12. val greeter = system.actorOf(Props[GreetingActor], name = "greeter")

  13. //向greeter发送信息，用Greeting来传递

  14. greeter ! Greeting("Charlie Parker")
复制代码



简直是无比强大啊，就这么几行代码就搞定了，接下来看你会更加震惊的。

我们回到Client类当中，找到ClientActor，它有两个方法，是之前说的preStart和receive方法，preStart方法用于连接master提交作业请求，receive方法用于接收从master返回的反馈信息。

我们先看preStart方法吧。


  1. override def preStart() = {

  2.     // 这里需要把master的地址转换成akka的地址，然后通过这个akka地址获得指定的actor

  3.     // 它的格式是"akka.tcp://%s@%s:%s/user/%s".format(systemName, host, port, actorName)

  4.     masterActor = context.actorSelection(Master.toAkkaUrl(driverArgs.master))

  5.     // 把自身设置成远程生命周期的事件

  6.     context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])

  7. 

  8.     driverArgs.cmd match {

  9.       case "launch" =>

  10.         // 此处省略100个字

  11.         val mainClass = "org.apache.spark.deploy.worker.DriverWrapper"

  12.         // 此处省略100个字

  13.         // 向master发送提交Driver的请求，把driverDescription传过去，RequestSubmitDriver前面说过了，是个case class

  14.         masterActor ! RequestSubmitDriver(driverDescription)

  15. 

  16.       case "kill" =>

  17.         val driverId = driverArgs.driverId

  18.         val killFuture = masterActor ! RequestKillDriver(driverId)

  19.     }

  20. }
复制代码



从上面的代码看得出来，它需要设置master的连接地址，最后提交了一个RequestSubmitDriver的信息。在receive方法里面，就是等待接受回应了，有两个Response分别对应着这里的launch和kill。

线索貌似到这里就断了，那下一步在哪里了呢？当然是在Master里面啦，怎么知道的，猜的，哈哈。

Master也是继承了Actor，在它的main函数里面找到了以下代码：

  1. val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf, 

  2. 　　securityManager = securityMgr)

  3. val actor = actorSystem.actorOf(Props(classOf[Master], host, boundPort, webUiPort, securityMgr), actorName)

  4. val timeout = AkkaUtils.askTimeout(conf)

  5. val respFuture = actor.ask(RequestWebUIPort)(timeout)

  6. val resp = Await.result(respFuture, timeout).asInstanceOf[WebUIPortResponse]
复制代码


和前面的actor基本一致，多了actor.ask这句话，查了一下官网的文档，这句话的意思的发送消息，并且接受一个Future作为response，和前面的actor ！ message的区别就是它还接受返回值。

具体的Akka的用法，大家还是参照官网吧，Akka确实如它官网所言的那样子，是一个简单、强大、并行的分布式框架。

小结：

Akka的使用确实简单，短短的几行代码即刻完成一个通信功能，比Socket简单很多。但是它也逃不脱我们常说的那些东西，请求、接收请求、传递的消息、注册的地址和端口这些概念。


调度schedule
我们接下来查找Master的receive方法吧，Master是作为接收方的，而不是主动请求，这点和hadoop是一致的。


  1.    case RequestSubmitDriver(description) => {

  2.         val driver = createDriver(description)

  3.         persistenceEngine.addDriver(driver)

  4.         waitingDrivers += driver

  5.         drivers.add(driver)

  6.         // 调度

  7.         schedule()

  8.          // 告诉client，提交成功了，把driver.id告诉它

  9.         sender ! SubmitDriverResponse(true, Some(driver.id), s"Driver successfully submitted as ${driver.id}")

  10.       }
复制代码



这里我们主要看schedule方法就可以了，它是执行调度的方法。


  1. private def schedule() {

  2.     if (state != RecoveryState.ALIVE) { return }

  3. 

  4.     // 首先调度Driver程序，从workers里面随机抽一些出来

  5.     val shuffledWorkers = Random.shuffle(workers) 

  6.     for (worker <- shuffledWorkers if worker.state == WorkerState.ALIVE) {

  7.       for (driver <- waitingDrivers) {

  8.         // 判断内存和cpu够不够，够的就执行了哈

  9.         if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {

  10.           launchDriver(worker, driver)

  11.           waitingDrivers -= driver

  12.         }

  13.       }

  14.     }

  15. 

  16.     // 这里是按照先进先出的，spreadOutApps是由spark.deploy.spreadOut参数来决定的，默认是true

  17.     if (spreadOutApps) {

  18.       // 遍历一下app

  19.       for (app <- waitingApps if app.coresLeft > 0) {

  20.         // canUse里面判断了worker的内存是否够用，并且该worker是否已经包含了该app的Executor

  21.         val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)

  22.           .filter(canUse(app, _)).sortBy(_.coresFree).reverse

  23.         val numUsable = usableWorkers.length

  24.         val assigned = new Array[Int](numUsable) 

  25.         // 记录每个节点的核心数

  26.         var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)

  27.         var pos = 0

  28.         // 遍历直到分配结束

  29.         while (toAssign > 0) {

  30.           // 从0开始遍历可用的work，如果可用的cpu减去已经分配的>0,就可以分配给它

  31.           if (usableWorkers(pos).coresFree - assigned(pos) > 0) {

  32.             toAssign -= 1

  33.             // 这个位置的work的可分配的cpu数+1

  34.             assigned(pos) += 1

  35.           }

  36.           pos = (pos + 1) % numUsable

  37.         }

  38.         // 给刚才标记的worker分配任务

  39.         for (pos <- 0 until numUsable) {

  40.           if (assigned(pos) > 0) {

  41.             val exec = app.addExecutor(usableWorkers(pos), assigned(pos))

  42.             launchExecutor(usableWorkers(pos), exec)

  43.             app.state = ApplicationState.RUNNING

  44.           }

  45.         }

  46.       }

  47.     } else {

  48.       // 这种方式和上面的方式的区别是，这种方式尽可能用少量的节点来完成这个任务

  49.       for (worker <- workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) {

  50.         for (app <- waitingApps if app.coresLeft > 0) {

  51.           // 判断条件是worker的内存比app需要的内存多

  52.           if (canUse(app, worker)) {

  53.             val coresToUse = math.min(worker.coresFree, app.coresLeft)

  54.             if (coresToUse > 0) {

  55.               val exec = app.addExecutor(worker, coresToUse)

  56.               launchExecutor(worker, exec)

  57.               app.state = ApplicationState.RUNNING

  58.             }

  59.           }

  60.         }

  61.       }

  62.     }

  63.   }
复制代码



它的调度器是这样的，先调度Driver程序，然后再调度App，调度App的方式是从各个worker的里面和App进行匹配，看需要分配多少个cpu。
那我们接下来看两个方法launchDriver和launchExecutor即可。


  1.   def launchDriver(worker: WorkerInfo, driver: DriverInfo) {

  2.     logInfo("Launching driver " + driver.id + " on worker " + worker.id)

  3.     worker.addDriver(driver)

  4.     driver.worker = Some(worker)

  5.     worker.actor ! LaunchDriver(driver.id, driver.desc)

  6.     driver.state = DriverState.RUNNING

  7.   }
复制代码



给worker发送了一个LaunchDriver的消息，下面在看launchExecutor的方法。


  1.   def launchExecutor(worker: WorkerInfo, exec: ExecutorInfo) {

  2.     logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)

  3.     worker.addExecutor(exec)

  4.     worker.actor ! LaunchExecutor(masterUrl,

  5.       exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory)

  6.     exec.application.driver ! ExecutorAdded(

  7.       exec.id, worker.id, worker.hostPort, exec.cores, exec.memory)

  8.   }
复制代码



它要做的事情多一点，除了给worker发送LaunchExecutor指令外，还需要给driver发送ExecutorAdded的消息，说你的任务已经有人干了。

在继续Worker讲之前，我们先看看它是怎么注册进来的，每个Worker启动之后，会自动去请求Master去注册自己，具体我们可以看receive的方法里面的RegisterWorker这一段，它需要上报自己的内存、Cpu、地址、端口等信息，注册成功之后返回RegisteredWorker信息给它，说已经注册成功了。


Worker执行
同样的，我们到Worker里面在receive方法找LaunchDriver和LaunchExecutor就可以找到我们要的东西。


  1. case LaunchDriver(driverId, driverDesc) => {

  2.       logInfo(s"Asked to launch driver $driverId")

  3.       val driver = new DriverRunner(driverId, workDir, sparkHome, driverDesc, self, akkaUrl)

  4.       drivers(driverId) = driver

  5.       driver.start()

  6. 

  7.       coresUsed += driverDesc.cores

  8.       memoryUsed += driverDesc.mem

  9. }
复制代码



看一下start方法吧，start方法里面，其实是new Thread().start()，run方法里面是通过传过来的DriverDescription构造的一个命令，丢给ProcessBuilder去执行命令，结束之后调用。

worker ！DriverStateChanged通知worker，worker再通过master ! DriverStateChanged通知master，释放掉worker的cpu和内存。

同理，LaunchExecutor执行完毕了，通过worker ! ExecutorStateChanged通知worker，然后worker通过master ! ExecutorStateChanged通知master，释放掉worker的cpu和内存。

下面我们再梳理一下这个过程，只包括Driver注册，Driver运行之后的过程在之后的文章再说，比较复杂。

1、Client通过获得Url地址获得ActorSelection（master的actor引用）,然后通过ActorSelection给Master发送注册Driver请求（RequestSubmitDriver）
2、Master接收到请求之后就开始调度了，从workers列表里面找出可以用的Worker
3、通过Worker的actor引用ActorRef给可用的Worker发送启动Driver请求（LaunchDriver）
4、调度完毕之后，给Client回复注册成功消息(SubmitDriverResponse)
5、Worker接收到LaunchDriver请求之后，通过传过来的DriverDescription的信息构造出命令来，通过ProcessBuilder执行
6、ProcessBuilder执行完命令之后，通过DriverStateChanged通过Worker
7、Worker最后把DriverStateChanged汇报给Master

后记：听超哥说，org.apache.spark.deploy.Client这个类快要被删除了，不知道cluster的这种模式是不是也被放弃了，官方给出来的例子推荐的是client模式->直接运行程序。难怪在作业调度的时候，看到别的actor叫driverActor。

不过这篇文章还有存在的意义, Akka和调度这块，和我现在正在写的第三篇以及第四篇关系很密切。


相关内容推荐：
Spark源码系列（一）spark-submit提交作业过程
Spark源码系列（二）RDD详解
Spark源码系列（三）作业运行过程
Spark源码系列（四）图解作业生命周期
Spark源码系列（五）分布式缓存
Spark源码系列（六）Shuffle的过程解析
Spark源码系列（七）Spark on yarn具体实现
Spark源码系列（八）Spark Streaming实例分析

作者：岑玉海


欢迎加入about云群90371779 、322273151、432264021 ，云计算爱好者群，亦可关注about云腾讯认证空间||关注本站微信
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
sunshine_junge

新手上路

关注

  • 80

    主题

  • 169

    帖子

  • 52

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000