spark streaming從1.2開始提供了資料的零丟失，想享受這個特性，需要滿足如下條件：

1. 資料輸入需要可靠的sources和可靠的receivers

2. 應用metadata必須通過應用driver checkpoint

3. WAL（write ahead log）

1.1. 可靠的sources和receivers

spark streaming可以通過多種方式作為資料sources（包括kafka），輸入資料通過receivers接收，通過replication儲存於spark中（為了faultolerance，預設複製到兩個spark executors），如果資料複製完成，receivers可以知道（例如kafka中更新offsets到zookeeper中）。這樣當receivers在接收資料過程中crash掉，不會有資料丟失，receivers沒有複製的資料，當receiver恢復後重新接收。

1.2. metadata checkpoint

可靠的sources和receivers，可以使資料在receivers失敗後恢復，然而在driver失敗後恢復是比較複雜的，一種方法是通過checkpoint metadata到HDFS或者S3。metadata包括：

· configuration

· code

· 一些排隊等待處理但沒有完成的RDD（僅僅是metadata，而不是data）

這樣當driver失敗時，可以通過metadata checkpoint，重構應用程式並知道執行到那個地方。

1.3. 資料可能丟失的場景

可靠的sources和receivers，以及metadata checkpoint也不可以保證資料的不丟失，例如：

· 兩個executor得到計算資料，並儲存在他們的記憶體中

· receivers知道資料已經輸入

· executors開始計算資料

· driver突然失敗

· driver失敗，那麼executors都會被kill掉

· 因為executor被kill掉，那麼他們記憶體中得資料都會丟失，但是這些資料不再被處理

· executor中的資料不可恢復

1.4. WAL

為了避免上面情景的出現，spark streaming 1.2引入了WAL。所有接收的資料通過receivers寫入HDFS或者S3中checkpoint目錄，這樣當driver失敗後，executor中資料丟失後，可以通過checkpoint恢復。

1.5. At-Least-Once

儘管WAL可以保證資料零丟失，但是不能保證exactly-once，例如下面場景：

· Receivers接收完資料並儲存到HDFS或S3

· 在更新offset前，receivers失敗了

· Spark Streaming以為資料接收成功，但是Kafka以為資料沒有接收成功，因為offset沒有更新到zookeeper

· 隨後receiver恢復了

· 從WAL可以讀取的資料重新消費一次，因為使用的kafka High-Level消費API，從zookeeper中儲存的offsets開始消費

1.6. WAL的缺點

通過上面描述，WAL有兩個缺點：

· 降低了receivers的效能，因為資料還要儲存到HDFS等分散式檔案系統

· 對於一些resources，可能存在重複的資料，比如Kafka，在Kafka中存在一份資料，在Spark Streaming也存在一份（以WAL的形式儲存在hadoop API相容的檔案系統中）

1.7. Kafka direct API

為了WAL的效能損失和exactly-once，spark streaming1.3中使用Kafka direct API。非常巧妙，Spark driver計算下個batch的offsets，指導executor消費對應的topics和partitions。消費Kafka訊息，就像消費檔案系統檔案一樣。

1. 不再需要kafka receivers，executor直接通過Kafka API消費資料

2. WAL不再需要，如果從失敗恢復，可以重新消費

3. exactly-once得到了保證，不會再從WAL中重複讀取資料

1.8. 總結

主要說的是spark streaming通過各種方式來保證資料不丟失，並保證exactly-once，每個版本都是spark streaming越來越穩定，越來越向生產環境使用發展。

sqoop import --connect jdbc:mysql://192.168.56.204:3306/sqoop --username hive --password hive --table jobinfo --target-dir /sqoop/test7 --inline-lob-limit 16777216 --fields-terminated-by '\t' -m 2

sqoop create-hive-table --connect jdbc:mysql://192.168.56.204:3306/sqoop --table jobinfo --username hive --password hive --hive-table sqtest --fields-terminated-by "\t" --lines-terminated-by "\n";

1、 spark應用提交後，經歷了一系列的轉換，最後成為task在每個節點上執行

2、 RDD的Action運算元觸發Job的提交，生成RDD DAG

3、 由DAGScheduler將RDD DAG轉化為Stage DAG，每個Stage中產生相應的Task集合

4、 TaskScheduler將任務分發到Executor執行

5、 每個任務對應相應的一個數據塊，只用使用者定義的函式處理資料塊

   通過org.apache.spark.deploy.Client類執行作業，作業執行命令如下：

作業執行流程描述：

1、客戶端提交作業給Master

2、Master讓一個Worker啟動Driver，即SchedulerBackend。Worker建立一個DriverRunner執行緒，DriverRunner啟動SchedulerBackend程序。

3、另外Master還會讓其餘Worker啟動Exeuctor，即ExecutorBackend。Worker建立一個ExecutorRunner執行緒，ExecutorRunner會啟動ExecutorBackend程序。

4、ExecutorBackend啟動後會向Driver的SchedulerBackend註冊。SchedulerBackend程序中包含DAGScheduler，它會根據使用者程式，生成執行計劃，並排程執行。對於每個stage的task，都會被存放到TaskScheduler中，ExecutorBackend向SchedulerBackend彙報的時候把TaskScheduler中的task排程到ExecutorBackend執行。

5、所有stage都完成後作業結束。

Driver執行在客戶端

作業執行流程描述：

1、客戶端啟動後直接執行使用者程式，啟動Driver相關的工作：DAGScheduler和BlockManagerMaster等。

2、客戶端的Driver向Master註冊。

3、Master還會讓Worker啟動Exeuctor。Worker建立一個ExecutorRunner執行緒，ExecutorRunner會啟動ExecutorBackend程序。

4、ExecutorBackend啟動後會向Driver的SchedulerBackend註冊。Driver的DAGScheduler解析作業並生成相應的Stage，每個Stage包含的Task通過TaskScheduler分配給Executor執行。

5、所有stage都完成後作業結束。