Spark Streaming事務

一：傳統事務概念

事務(Transaction)是訪問並可能更新數據庫中各類數據項的一個程序執行單元(unit)。事務一般由高級數據庫操縱語言或編程語言（如SQL，C++或Java）書寫的用戶程序的執行所引發，並用形如begin transaction和end transaction語句（或函數調用）來界定。事務由事務開始(begin transaction)和事務結束(end transaction)之間執行的全體操做組成。

例如：在關係數據庫中，一個事務能夠是一條SQL語句，一組SQL語句或整個程序。

 

特性：事務是恢復和併發控制的基本單位。

事務應該具備4個屬性：原子性、一致性、隔離性、持久性。這四個屬性一般稱爲ACID特性。

原子性（atomicity）。一個事務是一個不可分割的工做單位，事務中包括的諸操做要麼都作，要麼都不作。

一致性（consistency）。事務必須是使數據庫從一個一致性狀態變到另外一個一致性狀態。一致性與原子性是密切相關的。

隔離性（isolation）。一個事務的執行不能被其餘事務干擾。即一個事務內部的操做及使用的數據對併發的其餘事務是隔離的，併發執行的各個事務之間不能互相干擾。

持久性（durability）。持久性也稱永久性（permanence），指一個事務一旦提交，它對數據庫中數據的改變就應該是永久性的。接下來的其餘操做或故障不該該對其有任何影響。

二：spark Streaming中的事務

1. Exactly once容錯

　　2. 數據輸出不重複

一. 事務場景 :

　　以銀行轉賬一次爲例，A用戶轉帳給B用戶，如何保證事務的一致性，即A用戶可以轉出且只能轉出一次，B用戶可以收到且只能收到一次。　　

二.  Exactly once容錯：

　　事務處理中如何保證可以處理且只能處理一次，數據可以輸出且只能輸出一次。

　　數據丟失的主要場景以下：

　　　　在Receiver收到數據且經過Driver的調度，Executor開始計算數據的時候若是Driver忽然奔潰（致使Executor會被Kill掉），此時Executor會被Kill掉，那麼Executor中的數據就會丟失。

　1. 事務處理以下圖 :

事務處理過程解析 ： 

01.  InputStream ： 輸入數據 

02.  Executor ： 經過Receiver接收數據，當接收到數據後向Driver 彙報 

03.  Driver ： 經過StreamingContext接收到數據會啓動Job進行操做 

2.  解決事務源數據接收的安全性 ：

事務處理解析 ：

01.  Executor ： 在Receiver接收來自Kafka數據首先經過BlockManager寫入內存+磁盤或者經過WAL來保證數據的安全性；

02.  Executor  ： 經過Replication完成後產生Ack信號；

03.  Kafka ： 肯定收信息並讀取下一條數據，Kafka纔會進行updateOffsets操做 ；

04.  經過WAL機制讓全部的數據經過相似HDFS的方式進行安全性容錯處理，從而解決Executor被Kill掉後致使數據丟失能夠經過WAL機制恢復回來。

3.  解決Driver數據輸出的安全性 ：

數據的處理怎麼保證有且僅有被處理一次？

數據零丟失並不能保證Exactly Once，若是Receiver接收且保存起來後沒來得及更新updateOffsets時，就會致使數據被重複處理。

01.  經過StreamingContext接收數據經過CheckPoint進行容錯 ；

02. logging the updates ： 經過記錄跟蹤全部生成RDD的轉換（transformations）也就是記錄每一個RDD的lineage（血統）來從新計算生成丟失的分區數據 ；

 4.  Exactly Once的事務處理 ：

0一、 數據零丟失：必須有可靠的數據來源和可靠的Receiver，且整個應用程序的metadata必須進行checkpoint，且經過WAL來保證數據安全；

0二、Spark Streaming 1.3的時候爲了不WAL的性能損失和實現Exactly Once而提供了Kafka Direct API，把Kafka做爲文件存儲系統！！

0三、此時兼具備流的優點和文件系統的優點，Spark Streaming+Kafka就構建了完美的流處理世界！！！

0四、 數據不須要copy副本，不須要WAL性能損耗，不須要Receiver，全部的Executors直接經過kafka direct api直接消費數據，直接管理Offset，因此也不會重複消費數據；

三.   Spark Streaming數據輸出屢次重寫及解決方案：

　　一、 爲何會有這個問題，由於SparkStreaming在計算的時候基於SparkCore，SparkCore天生會作如下事情致使SparkStreaming的結果（部分）重複輸出:

一、Task重試；

二、慢任務推測；

三、Stage重複；

四、Job重試；

等會致使數據的丟失。

二、 對應的解決方案：

一、一個任務失敗就是job 失敗，設置spark.task.maxFailures次數爲1；

二、設置spark.speculation爲關閉狀態（由於慢任務推測其實很是消耗性能，因此關閉後能夠顯著的提升Spark Streaming處理性能）

三、Spark streaming on kafka的話，假如job失敗後能夠設置kafka的auto.offset.reset爲largest的方式會自動恢復job的執行。

最後再次強調: 能夠經過transform和foreachRDD基於業務邏輯代碼進行邏輯控制來實現數據不重複消費和輸出不重複！這二個方法相似於spark streaming的後門，能夠作任意想象的控制操做！