1.什么是事務(wù)？

? ? ? ?例如像銀行轉(zhuǎn)賬，A對(duì)B轉(zhuǎn)賬，B是否能收到多次轉(zhuǎn)賬，可能性不大；或者A轉(zhuǎn)給B的時(shí)候，A同樣費(fèi)用被扣了多次，B只收到一次，這樣也不可能。也就是說我們要做的事務(wù)級(jí)別的處理，簡(jiǎn)而言之這數(shù)據(jù)一定會(huì)被處理，且只被處理一次，能夠輸出且只能輸出一次。

2.Spark Streaming整個(gè)運(yùn)行角度的基本的情況

? ? ? spark streaming寫程序基于Driver和Executor兩部分，Driver的核心是StreamingContext，Receiver接收到的數(shù)據(jù)匯報(bào)給Driver（把元數(shù)據(jù)給Driver，而且Driver生產(chǎn)的RDD只對(duì)元數(shù)據(jù)感興趣），Driver為了數(shù)據(jù)安全進(jìn)行checkpoint（從數(shù)據(jù)角度講Block MeteData、DStreamGraph、Job），接下來(lái)在Executor上執(zhí)行，當(dāng)然也可能在多個(gè)Executor上執(zhí)行。

3.接收數(shù)據(jù)的角度講

? ? ? ?數(shù)據(jù)不斷流進(jìn)Executor（InputStream的產(chǎn)生是在Driver上的，屬于框架調(diào)度層面的，Executor中只有數(shù)據(jù)和RDD，實(shí)際上講也沒有所謂的RDD，只有怎么算這件事,InputStream：只是從邏輯層面上講）。數(shù)據(jù)流進(jìn)了receiver，不斷接受這個(gè)數(shù)據(jù)，為了保證這個(gè)數(shù)據(jù)安全性，默認(rèn)情況下把數(shù)據(jù)不斷通過容錯(cuò)方式進(jìn)行處理，容錯(cuò)方式進(jìn)行處理：寫進(jìn)磁盤，內(nèi)存同時(shí)有副本的形式，或者說wal。

? ? ? ?WAL機(jī)制：寫數(shù)據(jù)的時(shí)候，先通過WAL寫入文件系統(tǒng)中，然后在存儲(chǔ)到Executor，Executor存儲(chǔ)到內(nèi)存或磁盤中，這是storagelevel規(guī)定。假設(shè)前面沒寫成功，后面一定不會(huì)存儲(chǔ)到Executor中，不存儲(chǔ)到Executor中就不能匯報(bào)給Driver，這個(gè)數(shù)據(jù)不會(huì)被處理。

? ? ? 我們是否能一定確保數(shù)據(jù)的安全性呢？假如我有1G數(shù)據(jù)，在這次流的批次處理中需要處理，那我是否一定能處理這1G數(shù)據(jù)，其實(shí)不一定，wal確實(shí)能把要寫入磁盤的數(shù)據(jù)，就是進(jìn)行wal的數(shù)據(jù)，能夠保證它的安全，我們現(xiàn)在不考慮wal失敗的可能，wal失敗的可能不大，因?yàn)樗话銓?hdfs之類的。其實(shí)Executor接受數(shù)據(jù)是一條一條接收的（從流的角度講）或者說從一個(gè)對(duì)象一個(gè)對(duì)象接收的，他會(huì)把數(shù)據(jù)在內(nèi)存中，Receiver把數(shù)據(jù)積累到一定程度時(shí)候，才寫到wal或者寫到磁盤。還沒有積累到一定程度，Receiver（Executor）失敗了怎么辦，這時(shí)還是會(huì)有部分?jǐn)?shù)據(jù)丟失一點(diǎn)（是的）。談不到備份，因?yàn)檫€沒有準(zhǔn)備好數(shù)據(jù)塊，就是幾條數(shù)據(jù)

4.處理數(shù)據(jù)角度：

? ? ? 處理數(shù)據(jù)之前先checkpoint，checkpoint放到文件系統(tǒng)中，處理之后也會(huì)進(jìn)行checkpoint，在保存一下自己狀態(tài)。spark streaming內(nèi)部工作起來(lái)，絕對(duì)的核心是SparkContext；spark streaming就2點(diǎn)：就是StreamingContext，第一獲取數(shù)據(jù)，第二產(chǎn)生作業(yè)StreamingContext沒有解決執(zhí)行問題，解決執(zhí)行問還需要SparkContext；

? ? ? 假設(shè)出現(xiàn)崩潰的時(shí)候，需要數(shù)據(jù)恢復(fù)，從Driver的角度進(jìn)行恢復(fù)，Driver先checkpoint文件系統(tǒng)讀取進(jìn)來(lái)，而在內(nèi)部重新啟動(dòng)SparkContext。Driver里面恢復(fù)過數(shù)據(jù)，重新構(gòu)建StreamingContext，其實(shí)也是構(gòu)建SparkContext，恢復(fù)產(chǎn)生的元數(shù)據(jù)，再次產(chǎn)生RDD（恢復(fù)時(shí)候是基于上一次job或相對(duì)應(yīng)的job）再次提交到spark集群，在提交集群時(shí)候再次執(zhí)行，另外一方面包含了Receiver恢復(fù)，Receiver從新恢復(fù)在以前數(shù)據(jù)的基礎(chǔ)上接收數(shù)據(jù)，曾經(jīng)接受的數(shù)據(jù)它會(huì)通過wal之類的機(jī)制從磁盤重新恢復(fù)回來(lái)。

5.ExactlyOnce的事務(wù)處理：

1.數(shù)據(jù)零丟失：必須有可靠的數(shù)據(jù)來(lái)源和可靠的Receiver，且整個(gè)應(yīng)用程序的metadata必須進(jìn)行checkpoint，且通過wal來(lái)保證數(shù)據(jù)安全；

2.Spark?Streaming 1.3的時(shí)候?yàn)榱吮苊釽AL的性能損失和實(shí)現(xiàn)Exactly -once而提供了Kafka Direct API，把Kafka作為文件存儲(chǔ)系統(tǒng)！?。〈藭r(shí)兼具有流的優(yōu)勢(shì)和文件系統(tǒng)優(yōu)勢(shì)，至此，Spark Steaming + Kafka就構(gòu)建了完美的流處理世界?。?！所有的Executor通過KafkaAPI直接消費(fèi)數(shù)據(jù)，直接管理offset，所以也不會(huì)重復(fù)消費(fèi)數(shù)據(jù)；（此時(shí)可以保證數(shù)據(jù)一定會(huì)被處理且一定會(huì)被處理一次）事務(wù)實(shí)現(xiàn)啦?。?！

備注：

資料來(lái)源于：DT_大數(shù)據(jù)夢(mèng)工廠（Spark發(fā)行版本定制）

更多私密內(nèi)容，請(qǐng)關(guān)注微信公眾號(hào)：DT_Spark

如果您對(duì)大數(shù)據(jù)Spark感興趣，可以免費(fèi)聽由王家林老師每天晚上20：00開設(shè)的Spark永久免費(fèi)公開課，地址YY房間號(hào)：68917580