在大資料的各種框架中，hadoop無疑是大資料的主流，但是隨著電商企業的發展，hadoop只適用於一些離線資料的處理，無法應對一些實時資料的處理分析，我們需要一些實時計算框架來分析資料。因此出現了很多流式實時計算框架，比如Storm，Spark Streaming，Samaz等框架，本文主要講解Spark Streaming的工作原理以及如何使用。

一、流式計算

1.什麼是流？

Streaming：是一種資料傳送技術，它把客戶機收到的資料變成一個穩定連續的
流，源源不斷地送出，使使用者聽到的聲音或看到的圖象十分平穩，而且使用者在
整個檔案送完之前就可以開始在螢幕上瀏覽檔案。

2.常見的流式計算框架

• Apache Storm
• Spark Streaming
• Apache Samza

上述三種實時計算系統都是開源的分散式系統，具有低延遲、可擴充套件和容錯性
諸多優點，它們的共同特色在於：允許你在執行資料流程式碼時，將任務分配到
一系列具有容錯能力的計算機上並行執行。此外，它們都提供了簡單的API來
簡化底層實現的複雜程度。

二、Spark Streaming

1.Spark Streaming介紹

Spark Streaming是Spark生態系統當中一個重要的框架，它建立在Spark Core之上，下面這幅圖也可以看出Sparking Streaming在Spark生態系統中地位。

官方對於Spark Streaming的解釋如下：

Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. Data can be ingested from many sources like Kafka, Flume, Twitter, ZeroMQ, Kinesis or TCP sockets can be processed using complex algorithms expressed with high-level functions like map, reduce, join and window. Finally, processed data can be pushed out to filesystems, databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.

Spark Streaming是Spark Core的擴充套件應用，它具有可擴充套件，高吞吐量，對於流資料的可容錯性等特點。可以監控來自Kafka,Flumn,HDFS。Kinesis,Twitter，ZeroMQ或者Scoket套接字的資料通過複雜的演算法以及一系列的計算分析資料，並且可以將分析結果存入到HDFS檔案系統，資料庫以及前端頁面中。

Spark Streaming有以下特點：

• 高可擴充套件性，可以執行在上百臺機器上（Scales to hundreds of nodes）
• 低延遲，可以在秒級別上對資料進行處理（Achieves low latency）
• 高可容錯性（Efficiently recover from failures）
• 能夠整合平行計算程式，比如Spark Core（Integrates with batch and interactive processing）

2.Spark Streaming工作原理
對於Spark Core它的核心就是RDD，對於Spark Streaming來說，它的核心是DStream，DStream類似於RDD，它實質上一系列的RDD的集合，DStream可以按照秒數將資料流進行批量的劃分。首先從接收到流資料之後，將其劃分為多個batch，然後提交給Spark叢集進行計算，最後將結果批量輸出到HDFS或者資料庫以及前端頁面展示等等。可以參考下面這幅圖來幫助理解：

對於DStream如何理解呢？它是一系列連續的RDD，它是建立在Spark之上的不可變的，分散式資料集，在DStream中的每一個RDD包含著一定時間間隔的資料，如下圖所示：

那麼，Spark Streaming的工作原理是什麼呢？它是怎麼執行在叢集上的呢？其原理架構圖如下所示：

我們都知道Spark Core在初始化時會生成一個SparkContext物件來對資料進行後續的處理，相對應的Spark Streaming會建立一個Streaming Context，它的底層是SparkContext，也就是說它會將任務提交給SparkContext來執行，這也很好的解釋了DStream是一系列的RDD。當啟動Spark Streaming應用的時候，首先會在一個節點的Executor上啟動一個Receiver接受者，然後當從資料來源寫入資料的時候會被Receiver接收，接收到資料之後Receiver會將資料Split成很多個block，然後備份到各個節點（Replicate Blocks 容災恢復），然後Receiver向StreamingContext進行塊報告，說明資料在那幾個節點的Executor上，接著在一定間隔時間內StreamingContext會將資料處理為RDD並且交給SparkContext劃分到各個節點進行平行計算。

3.Spark Streaming Demo

介紹完Spark Streaming的基本原理之後，下面來看看如何執行Spark Streaming，官方給出了一個例子，從Socket源端監控收集資料執行wordcount的案例，案例很簡單，這裡不再說明，讀者可參考官方文件【http://spark.apache.org/docs/1.3.0/streaming-programming-guide.html】

對於Spark Streaming的程式設計模型有兩種方式

第一種：通過SparkConf來建立SparkStreaming

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ 
val conf=new SparkConf().setAppName("SparkStreamingDemo").setMaster("master")
val scc=new StreamingContext(conf,Seconds(1)) //每個1秒鐘檢測一次資料

第二種：通過SparkContext來建立，也就是在Spark-Shell命令列執行：

import org.apache.spark.streaming._
val scc=new StreamingContext(sc,Seconds(1))

當然，我們也可以收集來自HDFS檔案系統中資料，查閱Spark的原始碼，可以發現如下方法：

這個方法會監控指定HDFS檔案目錄下的資料，不過忽略以“.”開頭的檔案，也就是不會收集以“.”開頭的檔案進行資料的處理。

下面介紹一下如何從HDFS檔案系統上監控資料執行wordcount案例統計單詞數並且將結果打印出來的案例：

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ 

val ssc = new StreamingContext(sc, Seconds(5))

// read data
val lines = ssc.textFileStream("hdfs://hadoop-senior.shinelon.com:8020/user/shinelon/spark/streaming/")

// process
val words = lines.flatMap(_.split(" "))
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.print()

ssc.start()             // Start the computation
ssc.awaitTermination()  // Wait for the computation to terminate

上面程式會每個5秒鐘檢測一下HDFS檔案系統下的hdfs://hadoop-senior.shinelon.com:8020/user/shinelon/spark/streaming/目錄是否有新的資料，如果有就進行統計，然後將結果列印在控制檯。執行上面程式碼有兩種方式，可以執行Spark-shell客戶端後將上述命令一條條貼上到命令列執行，顯然這樣很麻煩；第二種就是將上面的程式寫入到一個指令碼檔案中載入到Spark-shell命令列中執行，這裡採用第二種方式：
在一個目錄下建立SparkStreamingDemo.scala檔案，內容如上面的程式碼所示。然後啟動Spark-shell客戶端。

$ bin/spark-shell --master local[2]

然後載入Spark Streaming應用：

scala>:load /opt/cdh-5.3.6/spark-1.3.0-bin-2.5.0-cdh5.3.6/SparkStreamingDemo.scala

然後上傳資料到上述HDFS檔案目錄下：

$ bin/hdfs dfs -put /opt/datas/wc.input /user/shinelon/spark/streaming/input7

該檔案內容如下所示：

hadoop hive
hadoop hbase
hadoop yarn
hadoop hdfs
hdfs spark

執行結果如下所示：

通常對於一個Spark Streaming的應用程式的編寫分下面幾步：

1. 定義一個輸入流源，比如獲取socket端的資料，HDFS,kafka中資料等等
2. 定義一系列的處理轉換操作，比如上面的map，reduce操作等等，Spark Streaming也有類似於SparkCore的transformation操作
3. 啟動程式收集資料（start()）
4. 等待程式停止（遇到錯誤終止或者手動停止awaitTermination()）
5. 手動終止應用程式（stop()）

可以使用saveAsTextFiles()方法將結果輸出到HDFS檔案系統上，讀者可以自行試驗將結果存入HDFS檔案系統中。

最後，介紹一下Spark Streaming應用程式開發的幾種常見方式：

1. Spark Shell Code：開發、測試（上面提到過，將程式碼一條條貼上到命令列執行，這種方式只適用於測試）
2. Spark Shell Load Scripts：開發、測試（編寫scala指令碼到spark-shell中執行）
3. IDE Develop App：開發、測試、打包JAR（生產環境），spark-submit提交應用程式