Java SparkStream WordCount

package stream;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * Created by xupan on 2017/11/27.
 */
public class SparkStreamOnline {

    public static void main(String[] args) {
        /**
         * 第一步：配置SparkConf
         * 1：至少兩條線程：由於Spark Stream運行的時候至少有一條線程不斷循環接收數據
         * 而且至少有一條線程用於處理接收數據，若是沒有線程處理數據，隨着時間的推移，內存和磁盤都會不堪重負
         * 2：對於集羣而言，每一個Executor不止一個線程，對於處理Spark Streaming的應用程序而言，每一個Executor
         * 通常分配多少core比較合適呢？通常5個左右的core是最佳的
         *
         **/
//        SparkConf conf = new SparkConf()
//                .setMaster("local[2]")
//                .setAppName("SparkStreamOnline");

        //運行在集羣環境不用指定core，能夠在提交的時候或則在默認配置【系統配置】中指定
        SparkConf conf = new SparkConf()
                .setMaster("spark://xupan001:7070")
                .setAppName("SparkStreamOnline");

        /**
         * 第二步：建立SparkStreamingContext，
         * 1：這SparkStreaming應用程序全部功能的起始點和程序調度的核心
         * JavaStreamingContext的構建基於SparkConf參數， 也能夠基於持久化的SparkStreamContext的內容恢復過來（典型的場景就是Driver崩潰後繼續上一次的狀態，
         * 此時狀態的恢復須要基於曾經的Checkpoint）
         * 2：在一個Spark Streaming程序中能夠建立若干個SparkStreamingContext對象：使用下一個JavaStreamingContext
         * 以前須要把前面一個SparkStreaming對象關閉掉，由此咱們得到一個重大的啓發SparkStreaming只是SparkCore上的一個
         * 應用程序，只不過SparkStreaming框架須要運行的話須要Spark工程師些業務程序處理代碼
         */
        JavaStreamingContext stream = new JavaStreamingContext(conf, Durations.seconds(5));//每隔5秒鐘

        /**
         *
         * 第三步：建立Spark Streaming輸入數據來源Input Stream：
         * 1:數據能夠是HDFS，File，KAFKA,Socket
         * 2:這裏數據來源於網絡端口，Spark Streaming 連接上該端口運行時一直監聽該端口的數據，該端口服務必須先存在，而且後續會根據業務須要不斷有數據產生
         * 3:若是常常間隔5秒沒數據的話不斷的啓動空Job，會形成空Job，會形成調度資源的浪費，由於並無數據須要進行運算，因此生產環境中提交Job前會判斷有無數據，若是沒有就不提交job
         **/
        JavaReceiverInputDStream lines = stream.socketTextStream("local", 9999);

        /**
         * 第四步：接下來就像對於RDD編程同樣基於DStream進行編程，緣由是DStreaming是RDD產生的模版或則說是類，在Spark Streaming具體發生計算前實質是把每一個Batch的DStreaming操做翻譯成RDD操做
         **/
        final JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            public Iterable<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" "));
            }
        });

        JavaPairDStream paris = words.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        });

        JavaPairDStream wordCount = paris.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        /**
         * Spark Streaming 要想運行具體的Job，DStream就必須有out put操做，
         * output stream 有不少觸發函數，如：print saveAsTextFile saveAsHadoopFiles等，
         * 最重要的一個方式是：foreachRDD,由於Spark Streaming處理的結果通常都會放在redis，DB，DashBoard上面
         *
         * 此處的print不會觸發job執行，由於如今一切都是在Spark Streaming控制下的
         * 對於Spark Streaming 而言job的觸發基於Durations時間間隔
         */
        wordCount.print();

        /**
         * Spark Streaming 執行引擎也就是Driver開始運行，Driver啓動的時候是位於一條新的線程中的，內部是由消息循環體用於接收
         * 應用程序自己或則Executor中的消息
         **/
        stream.start();

        stream.awaitTermination();

        stream.close();

    }
}