Spark Streaming部署應(yīng)用程序

Spark Streaming部署應(yīng)用程序

Requirements

運(yùn)行一個(gè)Spark Streaming應(yīng)用程序，有下面一些步驟

• 有管理器的集群-這是任何Spark應(yīng)用程序都需要的需求，詳見部署指南
• 將應(yīng)用程序打?yàn)閖ar包-你必須編譯你的應(yīng)用程序?yàn)閖ar包。如果你用spark-submit啟動(dòng)應(yīng)用程序，你不需要將Spark和Spark Streaming打包進(jìn)這個(gè)jar包。如果你的應(yīng)用程序用到了高級(jí)源（如kafka，flume），你需要將它們關(guān)聯(lián)的外部artifact以及它們的依賴打包進(jìn)需要部署的應(yīng)用程序jar包中。例如，一個(gè)應(yīng)用程序用到了TwitterUtils，那么就需要將spark-streaming-twitter_2.10以及它的所有依賴打包到應(yīng)用程序jar中。
• 為executors配置足夠的內(nèi)存-因?yàn)榻邮盏臄?shù)據(jù)必須存儲(chǔ)在內(nèi)存中，executors必須配置足夠的內(nèi)存用來保存接收的數(shù)據(jù)。注意，如果你正在做10分鐘的窗口操作，系統(tǒng)的內(nèi)存要至少能保存10分鐘的數(shù)據(jù)。所以，應(yīng)用程序的內(nèi)存需求依賴于使用它的操作。
• 配置checkpointing-如果stream應(yīng)用程序需要checkpointing，然后一個(gè)與Hadoop API兼容的容錯(cuò)存儲(chǔ)目錄必須配置為檢查點(diǎn)的目錄，流應(yīng)用程序?qū)heckpoint信息寫入該目錄用于錯(cuò)誤恢復(fù)。

• 配置應(yīng)用程序driver的自動(dòng)重啟-為了自動(dòng)從driver故障中恢復(fù)，運(yùn)行流應(yīng)用程序的部署設(shè)施必須能監(jiān)控driver進(jìn)程，如果失敗了能夠重啟它。不同的集群管理器，有不同的工具得到該功能

• Spark Standalone：一個(gè)Spark應(yīng)用程序driver可以提交到Spark獨(dú)立集群運(yùn)行，也就是說driver運(yùn)行在一個(gè)worker節(jié)點(diǎn)上。進(jìn)一步來看，獨(dú)立的集群管理器能夠被指示用來監(jiān)控driver，并且在driver失?。ɑ蛘呤怯捎诜橇愕耐顺龃a如exit(1)，或者由于運(yùn)行driver的節(jié)點(diǎn)的故障）的情況下重啟driver。
• YARN：YARN為自動(dòng)重啟應(yīng)用程序提供了類似的機(jī)制。

• Mesos： Mesos可以用Marathon提供該功能

• 配置write ahead logs-在Spark 1.2中，為了獲得極強(qiáng)的容錯(cuò)保證，我們引入了一個(gè)新的實(shí)驗(yàn)性的特性-預(yù)寫日志（write ahead logs）。如果該特性開啟，從receiver獲取的所有數(shù)據(jù)會(huì)將預(yù)寫日志寫入配置的checkpoint目錄。這可以防止driver故障丟失數(shù)據(jù)，從而保證零數(shù)據(jù)丟失。這個(gè)功能可以通過設(shè)置配置參數(shù)spark.streaming.receiver.writeAheadLogs.enable為true來開啟。然而，這些較強(qiáng)的語義可能以receiver的接收吞吐量為代價(jià)。這可以通過并行運(yùn)行多個(gè)receiver增加吞吐量來解決。另外，當(dāng)預(yù)寫日志開啟時(shí)，Spark中的復(fù)制數(shù)據(jù)的功能推薦不用，因?yàn)樵撊罩疽呀?jīng)存儲(chǔ)在了一個(gè)副本在存儲(chǔ)系統(tǒng)中。可以通過設(shè)置輸入DStream的存儲(chǔ)級(jí)別為StorageLevel.MEMORY_AND_DISK_SER獲得該功能。

升級(jí)應(yīng)用程序代碼

如果運(yùn)行的Spark Streaming應(yīng)用程序需要升級(jí)，有兩種可能的方法

• 啟動(dòng)升級(jí)的應(yīng)用程序，使其與未升級(jí)的應(yīng)用程序并行運(yùn)行。一旦新的程序（與就程序接收相同的數(shù)據(jù)）已經(jīng)準(zhǔn)備就緒，舊的應(yīng)用程序就可以關(guān)閉。這種方法支持將數(shù)據(jù)發(fā)送到兩個(gè)不同的目的地（新程序一個(gè)，舊程序一個(gè)）
• 首先，平滑的關(guān)閉（StreamingContext.stop(...)或JavaStreamingContext.stop(...)）現(xiàn)有的應(yīng)用程序。在關(guān)閉之前，要保證已經(jīng)接收的數(shù)據(jù)完全處理完。然后，就可以啟動(dòng)升級(jí)的應(yīng)用程序，升級(jí)的應(yīng)用程序會(huì)接著舊應(yīng)用程序的點(diǎn)開始處理。這種方法僅支持具有源端緩存功能的輸入源（如flume，kafka），這是因?yàn)楫?dāng)舊的應(yīng)用程序已經(jīng)關(guān)閉，升級(jí)的應(yīng)用程序還沒有啟動(dòng)的時(shí)候，數(shù)據(jù)需要被緩存。