《spark的容错分析》由会员分享，可在线阅读，更多相关《spark的容错分析（9页珍藏版）》请在金锄头文库上搜索。
1、Standalone部署的节点组成介绍 Spark的资料中对于RDD 这个概念涉及的比较多，但对于 RDD 如何运行起来， 如何对应到进程和线程的，着墨的不是很多。在实际的生产环境中，Spark总是会以集群的方式进行运行的，其中standalone的部署方式是所有集群方式中最为精简的一种，另外是Mesos和 YARN ，要理解其内部运行机理，显然要花更多的时间才能了解清楚。standalone cluster的组成standalone集群由三个不同级别的节点组成，分别是Master 主控节点， 可以类比为董事长或总舵主， 在整个集群之中， 最多只有一个Master处在 Active状态Work
2、er 工作节点， 这个是 manager,是分舵主， 在整个集群中， 可以有多个worker，如果 worker为零，什么事也做不了Executor 干苦力活的，直接受worker掌控，一个worker可以启动多个executor,启动的个数受限于机器中的cpu 核数这三种不同类型的节点各自运行于自己的JVM 进程之中 。Driver Application 提交到 standalone集群的应用程序称之为Driver Applicaton。Standalone集群启动及任务提交过 程详解上图总结了正常情况下Standalone集群的启动以及应用提交时，各节点之间有哪些消息交互。下面分集群启动
3、和应用提交两个过程来作详细说明。集群启动过程正常启动过程如下所述step 1: 启动 master $SPARK_HOME/sbin/start-master.sh step 2: 启动 worker ./bin/spark-class org.apache.spark.deploy.worker.Worker spark:/localhost:7077worker启动之后，会做两件事情1.将自己注册到Master, RegisterWorker 2.定期发送心跳消息给Master 任务提交过程step 1: 提交 application 利用如下指令来启动spark-shell MASTER
4、=spark:/127.0. 0.1 : 7077$SPARK_HOME/bin/spark-shell 运行 spark-shell时，会向Master发送 RegisterApplication请求日志位置： master运行产生的日志在$SPARK_HOME/logs目录下step 2: Master处理 RegisterApplication的请求之后收到 RegisterApplication请求之后， Mastet会做如下处理1.如果有 worker已经注册上来，发送LaunchExecutor指令给相应worker 2.如果没有，则什么事也不做step 3: 启动 Executo
5、r Worker在收到 LaunchExecutor指令之后，会启动Executor进程step 4: 注册 Executor 启动的 Executor进程会根据启动时的入参，将自己注册到Driver中的 SchedulerBackend 日志位置：executor的运行日志在$SPARK_HOME/work目录下step 5: 运行 Task SchedulerBackend收到 Executor的注册消息之后，会将提交到的Spark Job分解为多个具体的 Task ，然后通过LaunchTask指令将这些Task分散到各个Executor上真正的运行如果在调用runJob的时候，没有任何
6、的Executor注册到 SchedulerBackend，相应的处理逻辑是什么呢？1.SchedulerBackend会将 Task 存储在 TaskManager中2.一旦有 Executor注册上来，就将TaskManager管理的尚未运行的task提交到 executor中3.如果有多个job处于 pending状态，默认调度策略是FIFO ，即先提交的先运行测试步骤1.启动 Master 2.启动 spark-shell 3.执行 sc.textFile(“README.md“).count 4.启动 worker 5.注意 worker启动之后， spark-shell中打印出来的
7、日志消息Job 执行结束任务运行结束时，会将相应的Executor停掉。可以做如下的试验1.停止 spark-shell 2.利用 ps -ef|grep -i java查看 java进程，可以发现CoarseGrainedExecutorBackend进程已经退出小结通过上面的控制消息原语之间的先后顺序可以看出1.Master和 worker进程必须显式启动2.executor是被 worker隐式的带起3.集群的启动顺序1.Master必须先于其它节点启动2.worker和 driver哪个先启动，无所谓3.但 driver提交的 job 只有在有相应的worker注册到 Master之后
8、才可以被真正的执行 异常场景分析上面说明的是正常情况下，各节点的消息分发细节。那么如果在运行中，集群中的某些节点出现了问题，整个集群是否还能够正常处理Application中的任务呢？异常分析 1： worker异常退出在 Spark运行过程中，经常碰到的问题就是worker异常退出，当worker退出时，整个集群会有哪些故事发生呢? 请看下面的具体描述1.worker异常退出，比如说有意识的通过kill 指令将 worker杀死2.worker在退出之前，会将自己所管控的所有小弟executor全干掉3.worker需要定期向master改善心跳消息的， 现在 worker进程都已经玩完了，
9、 哪有心跳消息，所以 Master会在超时处理中意识到有一个“ 分舵 ” 离开了4.Master非常伤心，伤心的Master将情况汇报给了相应的Driver 5.Driver通过两方面确认分配给自己的Executor不幸离开了，一是Master发送过来的通知，二是 Driver没有在规定时间内收到Executor的 StatusUpdate，于是 Driver会将注册的Executor移除后果分析worker异常退出会带来哪些影响1.executor退出导致提交的task无法正常结束，会被再一次提交运行2.如果所有的worker都异常退出，则整个集群不可用3.需要有相应的程序来重启worker
10、进程，比如使用supervisord或 runit 测试步骤启动 Master 启动 worker 启动 spark-shell 手工 kill 掉 worker进程用 jps 或 ps -ef|grep -i java来查看启动着的java进程异常退出的代码处理定义于 ExecutorRunner.scala的 start函数def start() workerThread = new Thread( “ExecutorRunner for “ + fullId) overridedef run() fetchAndRunExecutor() workerThread.start() / S
11、hutdown hook that kills actors on shutdown.shutdownHook = new Thread() overridedef run() killProcess(Some(“Worker shutting down“) Runtime.getRuntime.addShutdownHook(shutdownHook) killProcess的过程就是停止相应CoarseGrainedExecutorBackend的过程。worker停止的时候，一定要先将自己启动的Executor停止掉。这是不是很像水浒中宋江的手段，李逵就是这样不明不白的把命给丢了。小结需
12、要特别指出的是，当 worker在启动 Executor的时候，是通过 ExecutorRunner来完成的，ExecutorRunner是一个独立的线程，和Executor是一对一的关系，这很重要。Executor作为一个独立的进程在运行，但会受到ExecutorRunner的严密监控。异常分析 2： executor异常退出Executor作为 Standalone集群部署方式下的最底层员工，一旦异常退出， 其后果会是什么呢？1.executor异常退出， ExecutorRunner注意到异常， 将情况通过ExecutorStateChanged汇报给 Master 2.Master收到
13、通知之后，非常不高兴，尽然有小弟要跑路，那还了得，要求Executor所属的worker再次启动3.Worker收到 LaunchExecutor指令，再次启动executor 作为一名底层员工，想轻易摞挑子不干是不成的。“人在江湖，身不由己“ 啊。测试步骤启动 Master 启动 Worker 启动 spark-shell 手工 kill 掉 CoarseGrainedExecutorBackend fetchAndRunExecutor fetchAndRunExecutor负责启动具体的Executor，并监控其运行状态，具体代码逻辑如下所示 def fetchAndRunExecuto
14、r() try / Create the executors working directoryval executorDir = new File(workDir, appId + “/“ + execId) if (!executorDir.mkdirs() thrownewIOException( “Failed to create directory “ + executorDir) / Launch the processval command = getCommandSeq logInfo(“Launch command: “ + command.mkString( “, “ “,
15、 “) val builder = new ProcessBuilder(command: _*).directory(executorDir) val env = builder.environment() for (key, value) logInfo(“Runner thread for executor “ + fullId + “ interrupted“) state = ExecutorState.KILLED killProcess(None) case e: Exception = logError(“Error running executor“, e) state =
16、ExecutorState.FAILED killProcess(Some(e.toString) 异常分析 3： master 异常退出worker和 executor异常退出的场景都讲到了，我们剩下最后一种情况了，master挂掉了怎么办？带头大哥如果不在了，会是什么后果呢？worker没有汇报的对象了，也就是如果executor再次跑飞， worker是不会将executor启动起来的，大哥没给指令无法向集群提交新的任务老的任务即便结束了，占用的资源也无法清除，因为资源清除的指令是Master发出的怎么样，知道后果很严重了吧？别看老大平时不干活，要真的不在，仅凭小弟们是不行的。Master单点失效问题的解决那么怎么解决Master单点失效的问题呢？你说再加一个Master就是了，两个老大。 两个老大如果同时具有指挥权，结果也将是灾难性的。设立一个副职人员，当目前的正职挂掉之后，副职接管。 也就是同一时刻，有且只有一个act
