spark容錯(cuò)分析.docx

Standalone部署的節(jié)點(diǎn)組成介紹Spark的資料中對于RDD這個(gè)概念涉及的比較多，但對于RDD如何運(yùn)行起來，如何對應(yīng)到進(jìn)程與線程的，著墨的不是很多。在實(shí)際的生產(chǎn)環(huán)境中，Spark總是會以集群的方式進(jìn)行運(yùn)行的，其中standalone的部署方式是所有集群方式中最為精簡的一種，另外是Mesos與YARN，要理解其內(nèi)部運(yùn)行機(jī)理，顯然要花更多的時(shí)間才能了解清楚。standalone cluster的組成standalone集群由三個(gè)不同級別的節(jié)點(diǎn)組成，分別是 Master主控節(jié)點(diǎn)，可以類比為董事長或總舵主，在整個(gè)集群之中，最多只有一個(gè)Master處在Active狀態(tài) Worker工作節(jié)點(diǎn) ，這個(gè)是manager,是分舵主， 在整個(gè)集群中，可以有多個(gè)worker，如果worker為零，什么事也做不了 Executor干苦力活的，直接受worker掌控，一個(gè)worker可以啟動多個(gè)executor,啟動的個(gè)數(shù)受限于機(jī)器中的cpu核數(shù)這三種不同類型的節(jié)點(diǎn)各自運(yùn)行于自己的JVM進(jìn)程之中。Driver Application提交到standalone集群的應(yīng)用程序稱之為Driver Applicaton。Standalone集群啟動及任務(wù)提交過程詳解上圖總結(jié)了正常情況下Standalone集群的啟動以及應(yīng)用提交時(shí)，各節(jié)點(diǎn)之間有哪些消息交互。下面分集群啟動與應(yīng)用提交兩個(gè)過程來作詳細(xì)說明。集群啟動過程正常啟動過程如下所述step 1: 啟動master$SPARK_HOME/sbin/start-master.shstep 2: 啟動worker./bin/spark-class org.apache.spark.deploy.worker.Worker spark:/localhost:7077worker啟動之后，會做兩件事情1. 將自己注冊到Master, RegisterWorker2. 定期發(fā)送心跳消息給Master任務(wù)提交過程step 1: 提交application利用如下指令來啟動spark-shellMASTER=spark:/127.0.0.1:7077 $SPARK_HOME/bin/spark-shell運(yùn)行spark-shell時(shí)，會向Master發(fā)送RegisterApplication請求日志位置：master運(yùn)行產(chǎn)生的日志在$SPARK_HOME/logs目錄下step 2: Master處理RegisterApplication的請求之后收到RegisterApplication請求之后，Mastet會做如下處理1. 如果有worker已經(jīng)注冊上來，發(fā)送LaunchExecutor指令給相應(yīng)worker2. 如果沒有，則什么事也不做step 3: 啟動ExecutorWorker在收到LaunchExecutor指令之后，會啟動Executor進(jìn)程step 4: 注冊Executor啟動的Executor進(jìn)程會根據(jù)啟動時(shí)的入?yún)?，將自己注冊到Driver中的SchedulerBackend日志位置： executor的運(yùn)行日志在$SPARK_HOME/work目錄下step 5: 運(yùn)行TaskSchedulerBackend收到Executor的注冊消息之后，會將提交到的Spark Job分解為多個(gè)具體的Task，然后通過LaunchTask指令將這些Task分散到各個(gè)Executor上真正的運(yùn)行如果在調(diào)用runJob的時(shí)候，沒有任何的Executor注冊到SchedulerBackend，相應(yīng)的處理邏輯是什么呢？1. SchedulerBackend會將Task存儲在TaskManager中2. 一旦有Executor注冊上來，就將TaskManager管理的尚未運(yùn)行的task提交到executor中3. 如果有多個(gè)job處于pending狀態(tài)，默認(rèn)調(diào)度策略是FIFO，即先提交的先運(yùn)行測試步驟1. 啟動Master2. 啟動spark-shell3. 執(zhí)行 sc.textFile(README.md).count4. 啟動worker5. 注意worker啟動之后，spark-shell中打印出來的日志消息Job執(zhí)行結(jié)束任務(wù)運(yùn)行結(jié)束時(shí)，會將相應(yīng)的Executor停掉?？梢宰鋈缦碌脑囼?yàn)1. 停止spark-shell2. 利用ps -ef|grep -i java查看java進(jìn)程，可以發(fā)現(xiàn)CoarseGrainedExecutorBackend進(jìn)程已經(jīng)退出小結(jié)通過上面的控制消息原語之間的先后順序可以看出1. Master與worker進(jìn)程必須顯式啟動2. executor是被worker隱式的帶起3. 集群的啟動順序1. Master必須先于其它節(jié)點(diǎn)啟動2. worker與driver哪個(gè)先啟動，無所謂3. 但driver提交的job只有在有相應(yīng)的worker注冊到Master之后才可以被真正的執(zhí)行異常場景分析上面說明的是正常情況下，各節(jié)點(diǎn)的消息分發(fā)細(xì)節(jié)。那么如果在運(yùn)行中，集群中的某些節(jié)點(diǎn)出現(xiàn)了問題，整個(gè)集群是否還能夠正常處理Application中的任務(wù)呢？異常分析1： worker異常退出在Spark運(yùn)行過程中，經(jīng)常碰到的問題就是worker異常退出，當(dāng)worker退出時(shí)，整個(gè)集群會有哪些故事發(fā)生呢? 請看下面的具體描述1. worker異常退出，比如說有意識的通過kill指令將worker殺死2. worker在退出之前，會將自己所管控的所有小弟executor全干掉3. worker需要定期向master改善心跳消息的，現(xiàn)在worker進(jìn)程都已經(jīng)玩完了，哪有心跳消息，所以Master會在超時(shí)處理中意識到有一個(gè)“分舵”離開了4. Master非常傷心，傷心的Master將情況匯報(bào)給了相應(yīng)的Driver5. Driver通過兩方面確認(rèn)分配給自己的Executor不幸離開了，一是Master發(fā)送過來的通知，二是Driver沒有在規(guī)定時(shí)間內(nèi)收到Executor的StatusUpdate，于是Driver會將注冊的Executor移除后果分析worker異常退出會帶來哪些影響1. executor退出導(dǎo)致提交的task無法正常結(jié)束，會被再一次提交運(yùn)行2. 如果所有的worker都異常退出，則整個(gè)集群不可用3. 需要有相應(yīng)的程序來重啟worker進(jìn)程，比如使用supervisord或runit測試步驟 啟動Master 啟動worker 啟動spark-shell 手工kill掉worker進(jìn)程 用jps或ps -ef|grep -i java來查看啟動著的java進(jìn)程異常退出的代碼處理定義于ExecutorRunner.scala的start函數(shù)def start() workerThread = new Thread(ExecutorRunner for + fullId) override def run() fetchAndRunExecutor() workerThread.start() / Shutdown hook that kills actors on shutdown. shutdownHook = new Thread() override def run() killProcess(Some(Worker shutting down) Runtime.getRuntime.addShutdownHook(shutdownHook) killProcess的過程就是停止相應(yīng)CoarseGrainedExecutorBackend的過程。worker停止的時(shí)候，一定要先將自己啟動的Executor停止掉。這是不是很像水滸中宋江的手段，李逵就是這樣不明不白的把命給丟了。小結(jié)需要特別指出的是，當(dāng)worker在啟動Executor的時(shí)候，是通過ExecutorRunner來完成的，ExecutorRunner是一個(gè)獨(dú)立的線程，與Executor是一對一的關(guān)系，這很重要。Executor作為一個(gè)獨(dú)立的進(jìn)程在運(yùn)行，但會受到ExecutorRunner的嚴(yán)密監(jiān)控。異常分析2： executor異常退出Executor作為Standalone集群部署方式下的最底層員工，一旦異常退出，其后果會是什么呢？1. executor異常退出，ExecutorRunner注意到異常，將情況通過ExecutorStateChanged匯報(bào)給Master2. Master收到通知之后，非常不高興，盡然有小弟要跑路，那還了得，要求Executor所屬的worker再次啟動3. Worker收到LaunchExecutor指令，再次啟動executor作為一名底層員工，想輕易摞挑子不干是不成的。人在江湖，身不由己“啊。測試步驟 啟動Master 啟動Worker 啟動spark-shell 手工kill掉CoarseGrainedExecutorBackendfetchAndRunExecutorfetchAndRunExecutor負(fù)責(zé)啟動具體的Executor，并監(jiān)控其運(yùn)行狀態(tài)，具體代碼邏輯如下所示def fetchAndRunExecutor() try / Create the executors working directory val executorDir = new File(workDir, appId + / + execId) if (!executorDir.mkdirs() throw new IOException(Failed to create directory + executorDir) / Launch the process val command = getCommandSeq logInfo(Launch command: + command.mkString(, , ) 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 = ExecutorState.FAILED killProcess(Some(e.toString) 異常分析3： master 異常退出worker與executor異常退出的場景都講到了，我們剩下最后一種情況了，master掛掉了怎么辦？帶頭大哥如果不在了，會是什么后果呢？ worker沒有匯報(bào)的對象了，也就是如果executor再次跑飛，worker是不會將executor啟動起來的，大哥沒給指令 無法向集群提交新的任務(wù) 老的任務(wù)即便結(jié)束了，占用的資源也無法清除，因?yàn)橘Y源清除的指令是Master發(fā)出的怎么樣，知道后果很嚴(yán)重了吧？別看老大平時(shí)不干活，要真的不在，僅憑小弟們是不行的。Master單點(diǎn)失效問題的解決那么怎么解決Master單點(diǎn)失效的問題呢？你說再加一個(gè)Master就是了，兩個(gè)老大。兩個(gè)老大如果同時(shí)具有指揮權(quán)，結(jié)果也將是災(zāi)難性的。設(shè)立一個(gè)副職人員，當(dāng)目前的正職掛掉之后，副職接管。也就是同一時(shí)刻，有且只有一個(gè)active master。注意不錯(cuò)，如何實(shí)現(xiàn)呢？使用zookeeper的ElectLeader功能，效果圖如下配置細(xì)節(jié)如何搭建zookeeper集群，這里不再廢話，哪天有空的話再整一整，或者可以參考寫的storm系列中談到的zookeeper的集群安裝步驟。假設(shè)zookeeper集群已經(jīng)設(shè)置成功，那么如何啟動standalone集群中的節(jié)點(diǎn)呢？有哪些特別的地方？conf/spark-env.sh在conf/spark-env.sh中，為SPARK_DAEMON_JAVA_OPTS添加如下選項(xiàng)System propertyMeaningspark.deploy.recoveryModeSet to ZOOKEEPER to enable standby Master recovery mode (default: NONE).spark.deploy.zookeeper.urlThe ZooKeeper cluster url (e.g., 192.168.1.100:2181,192.168.1.101:2181).spark.deploy.zookeeper.dirThe directory in ZooKeeper to store recovery state (default: /spark).設(shè)置SPARK_DAEMON_JAVA_OPTS的實(shí)際例子SPARK_DAE