Streaming job 的調(diào)度與執(zhí)行
結(jié)合文章 揭開Spark Streaming神秘面紗④ - job 的提交與執(zhí)行我們畫出了如下 job 調(diào)度執(zhí)行流程圖：
為什么很難保證 exactly once
上面這張流程圖最主要想說明的就是昌执，job 的提交執(zhí)行是異步的，與 checkpoint 操作并不是原子操作盒使。這樣的機(jī)制會(huì)引起數(shù)據(jù)重復(fù)消費(fèi)問題：
為了簡化問題容易理解留储，我們假設(shè)一個(gè) batch 只生成一個(gè) job，并且 spark.streaming.concurrentJobs 值為1董虱，該值代表 jobExecutor 線程池中線程的個(gè)數(shù)扼鞋，也即可以同時(shí)執(zhí)行的 job 的個(gè)數(shù)。
假設(shè)愤诱，batch duration 為2s云头，一個(gè) batch 的總共處理時(shí)間為1s，此時(shí)转锈，一個(gè) batch 開始了盘寡，第一步生成了一個(gè) job，假設(shè)花了0.1s撮慨，然后把該 job 丟到了 jobExecutor 線程池中等待調(diào)度執(zhí)行竿痰，由于 checkpoint 操作和 job 在線程池中執(zhí)行是異步的脆粥，在0.2s 的時(shí)候，checkpoint 操作完成并且此時(shí)開始了 job 的執(zhí)行影涉。
注意变隔，這個(gè)時(shí)候 checkpoint 完成了并且該 job 在 checkpoint 中的狀態(tài)是未完成的，隨后在第1s 的時(shí)候 job 完成了蟹倾，那么在這個(gè) batch 結(jié)束的時(shí)候 job 已經(jīng)完成了但該 job 在 checkpoint 中的狀態(tài)是未完成的（要了解 checkpoint 都保存了哪些數(shù)據(jù)請移步Spark Streaming的還原藥水——Checkpoint）匣缘。
在下一個(gè) batch 運(yùn)行到 checkpoint 之前就掛了（比如在拉取數(shù)據(jù)的時(shí)候掛了、OOM 掛了等等異常情況）鲜棠，driver 隨后從 checkpoint 中恢復(fù)肌厨，那么上述的 job 依然是未執(zhí)行的，根據(jù)使用的 api 不同豁陆，對于這個(gè) job 會(huì)再次拉取數(shù)據(jù)或從 wal 中恢復(fù)數(shù)據(jù)重新執(zhí)行該 job柑爸，那么這種情況下該 job 的數(shù)據(jù)就就會(huì)被重復(fù)處理。比如這時(shí)記次的操作盒音，那么次數(shù)就會(huì)比真實(shí)的多表鳍。
如果一個(gè) batch 有多個(gè) job 并且spark.streaming.concurrentJobs大于1，那么這種情況就會(huì)更加嚴(yán)重祥诽，因?yàn)檫@種情況下就會(huì)有多個(gè) job 已經(jīng)完成但在 checkpoint 中還是未完成狀態(tài)譬圣，在 driver 重啟后這些 job 對應(yīng)的數(shù)據(jù)會(huì)被重復(fù)消費(fèi)處理。
另一種會(huì)導(dǎo)致數(shù)據(jù)重復(fù)消費(fèi)的情況主要是由于 Spark 處理的數(shù)據(jù)單位是 partition 引起的雄坪。比如在處理某 partition 的數(shù)據(jù)到一半的時(shí)候厘熟，由于數(shù)據(jù)內(nèi)容或格式會(huì)引起拋異常，此時(shí) task 失敗诸衔，Spark 會(huì)調(diào)度另一個(gè)同樣的 task 執(zhí)行盯漂，那么此時(shí)引起 task 失敗的那條數(shù)據(jù)之前的該 partition 數(shù)據(jù)就會(huì)被重復(fù)處理颇玷，雖然這個(gè) task 被再次調(diào)度依然會(huì)失敗笨农。若是失敗還好，如果某些特殊的情況帖渠，新的 task 執(zhí)行成功了谒亦，那么我們就很難發(fā)現(xiàn)數(shù)據(jù)被重復(fù)消費(fèi)處理了。
如何保證 exactly once
至于如何才能保證 exactly once空郊，其實(shí)要根據(jù)具體情況而定（廢話）份招。總體來說狞甚，可以考慮以下幾點(diǎn)：
- 業(yè)務(wù)是否不能容忍即使是極少量的數(shù)據(jù)差錯(cuò)锁摔，如果是那么考慮 exactly once。如果可以容忍哼审，那就沒必要非實(shí)現(xiàn) exactly once 不可
- 即使重復(fù)處理極小部分?jǐn)?shù)據(jù)會(huì)不會(huì)對最終結(jié)果產(chǎn)生影響谐腰。若不會(huì)孕豹，那重復(fù)處理就重復(fù)吧，比如排重統(tǒng)計(jì)
- 若一定要保證 exactly once十气，應(yīng)該考慮將對 partition 處理和 checkpoint或自己實(shí)現(xiàn)類似 checkpoint 功能的操作做成原子的操作励背；并且對 partition 整批數(shù)據(jù)進(jìn)行類似事物的處理
