本博文內容主要包括以下幾點內容：
1、Spark Streaming on Polling from Flume實戰
2、Spark Streaming on Polling from Flume原始碼

一、推模式(Flume push SparkStreaming)與拉模式（SparkStreaming poll Flume）比較 ：

採用推模式：推模式的理解就是Flume作為快取，存有資料。監聽對應埠，如果服務可以連結，就將資料push過去。(簡單，耦合要低)，缺點是SparkStreaming 程式沒有啟動的話，Flume端會報錯，同時可能會導致Spark Streaming 程式來不及消費的情況。

採用拉模式：拉模式就是自己定義一個sink，SparkStreaming自己去channel裡面取資料，根據自身條件去獲取資料，穩定性好。

二、Flume poll 實戰：

1.Flume poll 配置

2、將下載後的三個jar包放入Flume安裝lib目錄：

3、配置Flume conf環境引數：
首先進入此入境

接下來在此檔案中的sink1中新增此內容：

agent1.sinks.sink1.type = org.apache.spark.streaming.flume.sink.SparkSink

agent1.sinks.sink1.hostname = Master

agent1.sinks
 
.sink1.port = 9999

agent1.sinks.sink1.channel = channel1

三、編寫程式碼：

public class SkarkStreamingPollDataFromFlume {
    public static void main(String[] args) {
         /*
        * 第一步：配置SparkConf：
        * 1，至少2條執行緒：因為Spark Streaming應用程式在執行的時候，至少有一條
        * 執行緒用於不斷的迴圈接收資料，並且至少有一條執行緒用於處理接受的資料（否則的話無法
        * 有執行緒用於處理資料，隨著時間的推移，記憶體和磁碟都會不堪重負）；
        * 2，對於叢集而言，每個Executor一般肯定不止一個Thread，那對於處理Spark Streaming的
        * 應用程式而言，每個Executor一般分配多少Core比較合適？根據我們過去的經驗，5個左右的
        * Core是最佳的（一個段子分配為奇數個Core表現最佳，例如3個、5個、7個Core等）；
        */
 


        SparkConf conf = new SparkConf().setAppName("SparkStreamingPollDataFromFlume").setMaster("local[2]");
          /*
         * 第二步：建立SparkStreamingContext：
         * 1，這個是SparkStreaming應用程式所有功能的起始點和程式排程的核心
         * SparkStreamingContext的構建可以基於SparkConf引數，也可基於持久化的SparkStreamingContext的內容
         * 來恢復過來（典型的場景是Driver崩潰後重新啟動，由於Spark Streaming具有連續7*24小時不間斷執行的特徵，
        * 所有需要在Driver重新啟動後繼續上衣系的狀態，此時的狀態恢復需要基於曾經的Checkpoint）；
         * 2，在一個Spark Streaming應用程式中可以建立若干個SparkStreamingContext物件，使用下一個SparkStreamingContext
         * 之前需要把前面正在執行的SparkStreamingContext物件關閉掉，由此，我們獲得一個重大的啟發SparkStreaming框架也只是
         * Spark Core上的一個應用程式而已，只不過Spark Streaming框架箱執行的話需要Spark工程師寫業務邏輯處理程式碼；
         */
        JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(30));
         /*

         * 第三步：建立Spark Streaming輸入資料來源input Stream：
         * 1，資料輸入來源可以基於File、HDFS、Flume、Kafka、Socket等
         * 2, 在這裡我們指定資料來源於網路Socket埠，Spark Streaming連線上該埠並在執行的時候一直監聽該埠
         *                  的資料（當然該埠服務首先必須存在）,並且在後續會根據業務需要不斷的有資料產生(當然對於Spark Streaming
         *                  應用程式的執行而言，有無資料其處理流程都是一樣的)；
         * 3,如果經常在每間隔5秒鐘沒有資料的話不斷的啟動空的Job其實是會造成排程資源的浪費，因為並沒有資料需要發生計算，所以
         *                 例項的企業級生成環境的程式碼在具體提交Job前會判斷是否有資料，如果沒有的話就不再提交Job；
         */                 

        JavaReceiverInputDStream lines = FlumeUtils.createPollingStream(jsc, "Master", 9999);
          /*
         * 第四步：接下來就像對於RDD程式設計一樣基於DStream進行程式設計！！！原因是DStream是RDD產生的模板（或者說類），在Spark Streaming具體
         * 發生計算前，其實質是把每個Batch的DStream的操作翻譯成為對RDD的操作！！！
         *對初始的DStream進行Transformation級別的處理，例如map、filter等高階函式等的程式設計，來進行具體的資料計算
   *    第4.1步：講每一行的字串拆分成單個的單詞
   */   
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<SparkFlumeEvent, String>() {

            @Override
            public Iterable<String> call(SparkFlumeEvent event) throws Exception {
                String line = new String(event.event().getBody().array());
                return Arrays.asList(line.split(" "));

            }
        });
          /*
         * 第四步：對初始的DStream進行Transformation級別的處理，例如map、filter等高階函式等的程式設計，來進行具體的資料計算
         * 第4.2步：在單詞拆分的基礎上對每個單詞例項計數為1，也就是word => (word, 1)
         */
        JavaPairDStream<String,Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                // TODO Auto-generated method stub
                return new Tuple2<String,Integer>(word,1);
            }
        });
          /*
         * 第四步：對初始的DStream進行Transformation級別的處理，例如map、filter等高階函式等的程式設計，來進行具體的資料計算
         * 第4.3步：在每個單詞例項計數為1基礎之上統計每個單詞在檔案中出現的總次數
         */
        JavaPairDStream<String,Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {

            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                // TODO Auto-generated method stub
                return v1 + v2;
            }
        });
         /*
         *        此處的print並不會直接出發Job的執行，因為現在的一切都是在Spark Streaming框架的控制之下的，對於Spark Streaming
         *        而言具體是否觸發真正的Job執行是基於設定的Duration時間間隔的
         *        諸位一定要注意的是Spark Streaming應用程式要想執行具體的Job，對Dtream就必須有output Stream操作，
         *        output Stream有很多型別的函式觸發，類print、saveAsTextFile、saveAsHadoopFiles等，最為重要的一個
         *        方法是foraeachRDD,因為Spark Streaming處理的結果一般都會放在Redis、DB、DashBoard等上面，foreachRDD
         *        主要就是用用來完成這些功能的，而且可以隨意的自定義具體資料到底放在哪裡！！！
         *
         */
        wordsCount.print();
          /*
         * Spark Streaming執行引擎也就是Driver開始執行，Driver啟動的時候是位於一條新的執行緒中的，當然其內部有訊息迴圈體，用於
         * 接受應用程式本身或者Executor中的訊息；
         */
        jsc.start();
        jsc.awaitTermination();
        jsc.close();
    }
}

啟動HDFS叢集：

啟動執行Flume：

啟動eclipse下的應用程式：

copy測試檔案hellospark.txt到Flume flume-conf.properties配置檔案中指定的/usr/local/flume/tmp/TestDir目錄下：

隔24秒後可以在eclipse程式控制臺中看到上傳的檔案單詞統計結果。

四：原始碼分析：

1、建立createPollingStream （FlumeUtils.scala ）：

2、引數配置：預設的全域性引數，private 級別配置無法修改：

3、建立FlumePollingInputDstream物件

4、繼承自ReceiverInputDstream並覆寫getReciver方法，呼叫FlumePollingReciver介面：

5、ReceiverInputDstream 構建了一個執行緒池，設定為後臺執行緒；並使用lazy和工廠方法建立執行緒和NioClientSocket（NioClientSocket底層使用NettyServer的方式）

6、receiverExecutor 內部也是執行緒池；connections是指連結分散式Flume叢集的FlumeConnection實體控制代碼的個數，執行緒拿到實體控制代碼訪問資料。

7、啟動時建立NettyTransceiver，根據並行度(預設5個)迴圈提交FlumeBatchFetcher

8、FlumeBatchFetcher run方法中從Receiver中獲取connection連結控制代碼ack跟訊息確認有關

9、獲取一批一批資料方法

補充說明：

使用Spark Streaming可以處理各種資料來源型別，如：資料庫、HDFS，伺服器log日誌、網路流，其強大超越了你想象不到的場景，只是很多時候大家不會用，其真正原因是對Spark、spark streaming本身不瞭解。

博文內容源自DT大資料夢工廠Spark課程。相關課程內容視訊可以參考：
百度網盤連結：http://pan.baidu.com/s/1slvODe1（如果連結失效或需要後續的更多資源，請聯絡QQ460507491或者微訊號：DT1219477246 獲取上述資料）。