點(diǎn)擊上方“IT那活兒”，關(guān)注后了解更多內(nèi)容，不管IT什么活兒，干就完了！?。?/span>

為了運(yùn)行一個(gè)spark streaming應(yīng)用程序，需要滿(mǎn)足以下條件 ：

1.1 使用集群管理器管理集群：

這是基本的要求。

1.2 打成jar包：

你必須將你的應(yīng)用程序編譯成jar包，使用spark-submit啟動(dòng)程序，然而如果你的程序使用的是高級(jí)數(shù)據(jù)源（例如kafka），你必須將kafka依賴(lài)打進(jìn)jar包。

1.3 為執(zhí)行節(jié)點(diǎn)配置足夠的內(nèi)存：

因?yàn)榻邮盏降臄?shù)據(jù)必須保存在內(nèi)存，所以執(zhí)行節(jié)點(diǎn)必須有足夠的內(nèi)存來(lái)存儲(chǔ)數(shù)據(jù)，如果要執(zhí)行10分鐘的窗口操作，系統(tǒng)必須在內(nèi)存中保留至少10分鐘的數(shù)據(jù)，因此應(yīng)用程序的內(nèi)存需求取決于其中使用的操作。

1.4 配置檢查點(diǎn)：

如果流應(yīng)用程序需要，則必須將Hadoop API兼容容錯(cuò)存儲(chǔ)（例如HDFS、S3等）中的目錄配置為檢查點(diǎn)目錄，并且流應(yīng)用程序的寫(xiě)入方式應(yīng)確保檢查點(diǎn)信息可用于故障恢復(fù)。

1.5 配置應(yīng)用驅(qū)動(dòng)程序的的自動(dòng)重啟：

為了從驅(qū)動(dòng)程序故障中自動(dòng)修復(fù)，用于運(yùn)行流應(yīng)用程序的部署基礎(chǔ)結(jié)構(gòu)必須監(jiān)視驅(qū)動(dòng)程序進(jìn)程，并在驅(qū)動(dòng)程序失敗時(shí)重新啟動(dòng)驅(qū)動(dòng)程序。不同的集群管理器有不同的工具來(lái)實(shí)現(xiàn)這一點(diǎn)。

• spark standalone：可以提交spark程序以以spark standalone方式運(yùn)行，也就是說(shuō)應(yīng)用程序在一個(gè)節(jié)點(diǎn)運(yùn)行，而且可以指示standalone集群管理器監(jiān)督驅(qū)動(dòng)程序，如果驅(qū)動(dòng)程序由于非零退出代碼或運(yùn)行驅(qū)動(dòng)程序的節(jié)點(diǎn)故障而失敗，則重新啟動(dòng)它。

• YARN：YARN支持自動(dòng)重啟應(yīng)用程序的類(lèi)似機(jī)制。

•  Mesos：Marathon已經(jīng)被用來(lái)實(shí)現(xiàn)這一目標(biāo)。

1.6 配置預(yù)寫(xiě)日志（write-ahead logs）：

自spark1.2，我們已經(jīng)引入了預(yù)寫(xiě)日志以實(shí)現(xiàn)強(qiáng)大的容錯(cuò)保證，如果啟用它，所有從receiver接收到的數(shù)據(jù)都會(huì)寫(xiě)入配置檢查點(diǎn)目錄中的預(yù)寫(xiě)日志。

這可以防止驅(qū)動(dòng)程序恢復(fù)時(shí)的數(shù)據(jù)丟失，從而確保零數(shù)據(jù)丟失，可以通過(guò)設(shè)置spark.streaming.receiver.writeAheadLog.enable=true來(lái)啟用它，然而這可能以單個(gè)接收器的接收吞吐為代價(jià)，但是這可以通過(guò)并行運(yùn)行更多接收器來(lái)彌補(bǔ)。

此外，建議在啟用預(yù)寫(xiě)日志時(shí)禁用spark內(nèi)接收數(shù)據(jù)的復(fù)制，因?yàn)樵撊罩疽汛鎯?chǔ)在已復(fù)制的存儲(chǔ)系統(tǒng)中，這可以通過(guò)設(shè)置存儲(chǔ)級(jí)別為StorageLevel.MEMORY_AND_DISK_SER來(lái)實(shí)現(xiàn)，使用S3（或任何不支持刷新的文件系統(tǒng)）進(jìn)行預(yù)寫(xiě)日志時(shí)，請(qǐng)記住啟用：

spark.streaming.driver.writeAheadLog.closeFileAfterWrite
和
spark.streaming.receiver.writeAheadLog.closeFileAfterWrite。

1.7 設(shè)置最大接收速率：

如果集群資源不夠大，spark streaming應(yīng)用程序無(wú)法以接收數(shù)據(jù)的速度處理數(shù)據(jù)，則可以通過(guò)設(shè)置記錄的最大速率限制來(lái)限制接收器的速率，請(qǐng)參閱：

• 接收器的配置參數(shù)spark.streaming.receiver.maxRate

• 直接kafka方法的配置參數(shù)spark.streaming.kafka.maxRatePerPartition

在Spark 1.5中，我們引入了一種稱(chēng)為背壓的功能，它消除了設(shè)置此速率限制的需要，因?yàn)镾park Streaming會(huì)自動(dòng)計(jì)算速率限制，并在處理?xiàng)l件發(fā)生變化時(shí)動(dòng)態(tài)調(diào)整速率限制?？赏ㄟ^(guò)將配置參數(shù)spark.streaming.backpressure.enabled設(shè)置為true來(lái)啟用此背壓。

• processing time：處理每個(gè)批次花費(fèi)的時(shí)間
• Scheduling Delay：批在隊(duì)列里等待前一批處理完成的時(shí)間

• 通過(guò)高效地使用群集資源，減少每批數(shù)據(jù)的處理時(shí)間。
• 設(shè)置正確的批大小，以便可以在接收數(shù)據(jù)時(shí)盡快處理數(shù)據(jù)批（即，數(shù)據(jù)處理與數(shù)據(jù)攝取保持同步）。

• 輸入數(shù)據(jù)（input data）：
  默認(rèn)情況下，通過(guò)接收器接收的輸入數(shù)據(jù)存儲(chǔ)在具有StorageLevel.memory_DISK_SER_2的執(zhí)行器內(nèi)存中。也就是說(shuō)，數(shù)據(jù)序列化為字節(jié)以減少GC開(kāi)銷(xiāo)，并復(fù)制以容忍執(zhí)行器故障。
  此外，數(shù)據(jù)首先保存在內(nèi)存中，只有當(dāng)內(nèi)存不足以保存流計(jì)算所需的所有輸入數(shù)據(jù)時(shí)，才會(huì)溢出到磁盤(pán)。這種序列化顯然有開(kāi)銷(xiāo)——接收器必須對(duì)接收到的數(shù)據(jù)進(jìn)行反序列化，并使用Spark的序列化格式對(duì)其重新序列化。
• 流操作生成的持久化RDD：
  流式計(jì)算生成的RDD可以保存在內(nèi)存中。例如，窗口操作將數(shù)據(jù)保存在內(nèi)存中，因?yàn)樗鼈儗⒈欢啻翁幚?。但是，與Spark Core默認(rèn)的StorageLevel.MEMORY_ONLY不同，流式計(jì)算生成的持久化RDD默認(rèn)使用StorageLevel.MEMORY_ONLY_DISK（即序列化）持久化，以最小化GC開(kāi)銷(xiāo)。

• 數(shù)據(jù)流的存儲(chǔ)級(jí)別：
  imput data數(shù)據(jù)和RDD在默認(rèn)情況下作為序列化字節(jié)持久化。與反序列化持久化相比，這減少了內(nèi)存使用和GC開(kāi)銷(xiāo)。啟用Kryo序列化進(jìn)一步減少了序列化大小和內(nèi)存使用。
• 清除舊數(shù)據(jù)：
  默認(rèn)情況下，由數(shù)據(jù)流轉(zhuǎn)換生成的所有輸入數(shù)據(jù)和持久化RDD將自動(dòng)清除。Spark Streaming根據(jù)所使用的轉(zhuǎn)換決定何時(shí)清除數(shù)據(jù)。例如，如果使用10分鐘的窗口操作，則Spark Streaming將保留最后10分鐘的數(shù)據(jù)，并主動(dòng)丟棄較舊的數(shù)據(jù)。通過(guò)設(shè)置streamingContext.remember，數(shù)據(jù)可以保留更長(zhǎng)的時(shí)間。
• CMS垃圾收集器：
  強(qiáng)烈建議使用并發(fā)標(biāo)記和掃描GC，以保持GC相關(guān)暫停始終較低。盡管已知并發(fā)GC會(huì)降低系統(tǒng)的總體處理吞吐量，但仍建議使用它來(lái)實(shí)現(xiàn)更一致的批處理時(shí)間。確保在驅(qū)動(dòng)程序（使用spark submit中的--driver java選項(xiàng)）和執(zhí)行器（使用spark配置spark.executor.extraJavaOptions）上設(shè)置CMS GC。

• RDD是一個(gè)不可變的、確定的和可重新計(jì)算的分布式數(shù)據(jù)集，每個(gè)RDD都會(huì)記住創(chuàng)建數(shù)據(jù)集的依賴(lài)。
• 如果RDD的任何分區(qū)由于工作節(jié)點(diǎn)故障而丟失，則可以使用操作依賴(lài)關(guān)系從原始容錯(cuò)數(shù)據(jù)集重新計(jì)算該分區(qū)。
• 假設(shè)所有的RDD轉(zhuǎn)換都是確定性的，那么不管Spark集群中是否出現(xiàn)故障，最終轉(zhuǎn)換的RDD中的數(shù)據(jù)都將始終相同。

• 接收和復(fù)制的數(shù)據(jù)—此數(shù)據(jù)在單個(gè)工作節(jié)點(diǎn)發(fā)生故障時(shí)仍然有效，因?yàn)樗母北敬嬖谟诹硪粋€(gè)節(jié)點(diǎn)上。
• 已接收但已緩沖用于復(fù)制的數(shù)據(jù)—由于未復(fù)制此數(shù)據(jù)，因此恢復(fù)此數(shù)據(jù)的唯一方法是再次從源獲取它。

• 運(yùn)行executors的任何工作節(jié)點(diǎn)都可能失敗，并且這些節(jié)點(diǎn)上的所有內(nèi)存中數(shù)據(jù)都將丟失。如果任何接收器在發(fā)生故障的節(jié)點(diǎn)上運(yùn)行，則其緩沖數(shù)據(jù)將丟失。
• 如果運(yùn)行Spark Streaming應(yīng)用程序的驅(qū)動(dòng)程序節(jié)點(diǎn)出現(xiàn)故障，則SparkContext顯然會(huì)丟失，所有executor及其內(nèi)存中的數(shù)據(jù)也會(huì)丟失。

• 最多一次：每條記錄要么處理一次，要么根本不處理。
• 至少一次：每條記錄將被處理一次或多次。這比最多一次強(qiáng)，因?yàn)樗_保不會(huì)丟失任何數(shù)據(jù)。但也可能有重復(fù)。
• 只有一次：每條記錄將精確處理一次-不會(huì)丟失任何數(shù)據(jù)，也不會(huì)多次處理任何數(shù)據(jù)。這顯然是三者中最有力的保證。

• 接收數(shù)據(jù)：使用接收器或其他方式從源接收數(shù)據(jù)。
• 轉(zhuǎn)換數(shù)據(jù)：使用數(shù)據(jù)流和RDD轉(zhuǎn)換處理接收的數(shù)據(jù)。
• 輸出數(shù)據(jù)：最終轉(zhuǎn)換的數(shù)據(jù)輸出到外部系統(tǒng)，如文件系統(tǒng)、數(shù)據(jù)庫(kù)、儀表板等。

• 接收數(shù)據(jù)：不同的輸入源提供不同的保證，文章后面將詳細(xì)討論。
• 轉(zhuǎn)換數(shù)據(jù)：由于RDD提供的保證，所有接收到的數(shù)據(jù)都將被精確地處理一次。即使出現(xiàn)故障，只要接收到的輸入數(shù)據(jù)是可訪問(wèn)的，最終轉(zhuǎn)換的RDD將始終具有相同的內(nèi)容。
• 推出數(shù)據(jù)：默認(rèn)情況下，輸出操作至少確保一次語(yǔ)義，因?yàn)樗Q于輸出操作的類(lèi)型（冪等式或非冪等式）和下游系統(tǒng)的語(yǔ)義（是否支持事務(wù)）。但用戶(hù)可以實(shí)現(xiàn)自己的事務(wù)機(jī)制，以實(shí)現(xiàn)精確的一次性語(yǔ)義。文章后面將詳細(xì)討論這一點(diǎn)。

• 可靠接收器-這些接收器僅在確保已復(fù)制接收到的數(shù)據(jù)后才確認(rèn)可靠來(lái)源。如果這樣的接收器出現(xiàn)故障，源將不會(huì)收到緩沖（未復(fù)制）數(shù)據(jù)的確認(rèn)。
  因此，如果接收器重新啟動(dòng)，源將重新發(fā)送數(shù)據(jù)，并且不會(huì)因故障而丟失任何數(shù)據(jù)。
• 不可靠的接收器-此類(lèi)接收器不發(fā)送確認(rèn)，因此在由于工作人員或驅(qū)動(dòng)程序故障而失敗時(shí)可能會(huì)丟失數(shù)據(jù)。

• 冪等更新：多次嘗試總是寫(xiě)入相同的數(shù)據(jù)。例如，saveAs***文件總是將相同的數(shù)據(jù)寫(xiě)入生成的文件。
• 事務(wù)性更新：所有更新都是以事務(wù)方式進(jìn)行的，因此更新只以原子方式進(jìn)行一次。實(shí)現(xiàn)這一點(diǎn)的一種方法是：