1，Exactly once 事務

什麼事Exactly once 事務？

資料僅處理一次並且僅輸出一次，這樣才是完整的事務處理。

Spark在執行出錯時不能保證輸出也是事務級別的。在Task執行一半的時候出錯了，雖然在語義上做了事務處理，資料僅被處理一次，但是如果是輸出到資料庫中，那有空能將結果多次儲存到資料庫中。Spark在任務失敗時會進行重試，這樣會導致結果多次儲存到資料庫中。

如下圖，當執行在Executor上的Receiver接收到資料通過BlockManager寫入記憶體和磁碟，或者通過WAL機制寫記錄日誌，然後把metedata資訊彙報給Driver。在Driver端定期進行checkpoint

Exactly once 事務的處理：

1，資料零丟失：必須有可靠的資料來源和可靠的Receiver，且整個應用程式的metadata必須進行checkpoint，且通過WAL來保證資料安全。

我們以資料來自Kafka為例，執行在Executor上的Receiver在接收到來自Kafka的資料時會向Kafka傳送ACK確認收到資訊並讀取下一條資訊，kafka會updateOffset來記錄Receiver接收到的偏移，這種方式保證了在Executor資料零丟失。

在Driver端，定期進行checkpoint操作，出錯時從Checkpoint

那麼資料可能丟失的地方有哪些呢和相應的解決方式？

    在Receiver收到資料且通過Driver的排程Executor開始計算資料的時候，如果Driver突然奔潰，則此時Executor會被殺死，那麼Executor中的資料就會丟失(如果沒有進行WAL的操作)。

解決方式：此時就必須通過例如WAL的方式，讓所有的資料都通過例如HDFS的方式首先進行安全性容錯處理。此時如果Executor

這種方式的弊端是通過WAL的方式會極大額損傷SparkStreaming中Receivers接收資料的效能。

資料重複讀取的情況：

    在Receiver收到資料儲存到HDFS等持久化引擎但是沒有來得及進行updateOffsets(以Kafka為例)，此時Receiver崩潰後重新啟動就會通過管理Kafka的Zookeeper中元資料再次重複讀取資料，但是此時SparkStreaming認為是成功的，但是kafka認為是失敗的(因為沒有更新offset到ZooKeeper中)，此時就會導致資料重新消費的情況。

    解決方式：以Receiver基於ZooKeeper的方式，當讀取資料時去訪問Kafka的元資料資訊，在處理程式碼中例如foreachRDD或transform時，將資訊寫入到記憶體資料庫中(memorySet)，在計算時讀取記憶體資料庫資訊，判斷是否已處理過，如果以處理過則跳過計算。這些元資料資訊可以儲存到記憶體資料結構或者memsql，sqllite中。

如果通過Kafka作為資料來源的話，Kafka中有資料，然後Receiver接收的時候又會有資料副本，這個時候其實是儲存資源的浪費。

Spark在1.3的時候為了避免WAL的效能損失和實現Exactly Once而提供了Kafka Direct API，把Kafka作為檔案儲存系統。此時兼具有流的優勢和檔案系統的優勢，至此Spark Streaming+Kafka就構建了完美的流處理世界（1，資料不需要拷貝副本；2，不需要WAL對效能的損耗；3，Kafka使用ZeroCopy比HDFS更高效）。所有的Executors通過Kafka API直接訊息資料，直接管理Offset，所以也不會重複消費資料。

2，輸出不重複

關於Spark Streaming資料輸出多次重寫及其解決方案：

1，為什麼會有這個問題，因為Spark Streaming在計算的時候基於Spark Core天生會做以下事情導致Spark Streaming的結果(部分)重複輸出。Task重試，慢任務推測，Stage重試，Job重試。

2，具體解決方案：

設定spark.task.maxFailures次數為1，這樣就不會有Task重試了。設定spark.speculation為關閉狀態，就不會有慢任務推測了，因為慢任務推測非常消耗效能，所以關閉後可以顯著提高Spark Streaming處理效能。

Spark Streaming On Kafka的話，Job失敗後可以設定Kafka的引數auto.offset.reset為largest方式。

    最後再次強調可以通過transform和foreachRDD基於業務邏輯程式碼進行邏輯控制來實現資料不重複消費和輸出不重複。這兩個方法類似於Spark Streaming的後門，可以做任意想象的控制操作。