/**
 * 1、local的模拟线程数必须大于等于2 因为一条线程被receiver(接受数据的线程)占用，另外一个线程是job执行
 * 2、Durations时间的设置，就是我们能接受的延迟度，这个我们需要根据集群的资源情况以及监控每一个job的执行时间来调节出最佳时间。
 * 3、 创建JavaStreamingContext有两种方式 （sparkconf、sparkcontext）
 * 4、业务逻辑完成后，需要有一个output operator
 * 5、JavaStreamingContext.start()straming框架启动之后是不能在次添加业务逻辑
 * 6、JavaStreamingContext.stop()无参的stop方法会将sparkContext一同关闭，stop(false) ,默认为true，会一同关闭
 * 7、JavaStreamingContext.stop()停止之后是不能在调用start   
 */

/**
 * foreachRDD  算子注意：
 * 1.foreachRDD是DStream中output operator类算子
 * 2.foreachRDD可以遍历得到DStream中的RDD，可以在这个算子内对RDD使用RDD的Transformation类算子进行转化，但是一定要使用rdd的Action类算子触发执行。
 * 3.foreachRDD可以得到DStream中的RDD，在这个算子内，RDD算子外执行的代码是在Driver端执行的，RDD算子内的代码是在Executor中执行。
 *
 */
public class WordCountOnline {
    @SuppressWarnings("deprecation")
    public static void main(String[] args) {
         
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordCountOnline");
        /**
         * 在创建streaminContext的时候 设置batch Interval
         */
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
//        JavaSparkContext sc = new JavaSparkContext(conf);
//        JavaStreamingContext jsc = new JavaStreamingContext(sc,Durations.seconds(5));
//        JavaSparkContext sparkContext = jsc.sparkContext();
        
        JavaReceiverInputDStream<String> lines = jsc.socketTextStream("node5", 9999);
        
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            /**
             * 
             */
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<String> call(String s) {
                return Arrays.asList(s.split(" "));
            }
        });

        JavaPairDStream<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
            /**
             * 
             */
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, Integer> call(String s) {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairDStream<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
            /**
             * 
             */
            private static final long serialVersionUID = 1L;

            @Override
            public Integer call(Integer i1, Integer i2) {
                return i1 + i2;
            }
        });
         
        //outputoperator类的算子   
//         counts.print();
//         counts.foreachRDD(new VoidFunction<JavaPairRDD<String,Integer>>() {
//
//            /**
//             * 
//             */
//            private static final long serialVersionUID = 1L;
//
//            @Override
//            public void call(JavaPairRDD<String, Integer> pairRDD) throws Exception {
//                
　　　　　　　　　　　//下面这行代码为action算子
//                pairRDD.foreach(new VoidFunction<Tuple2<String,Integer>>() {
//
//                    /**
//                     * 
//                     */
//                    private static final long serialVersionUID = 1L;
//
//                    @Override
//                    public void call(Tuple2<String, Integer> tuple)
//                            throws Exception {
//                        System.out.println("tuple ---- "+tuple );
//                    }
//                });
//            }
//        });
         jsc.start();
         //等待spark程序被终止
         jsc.awaitTermination();
         jsc.stop(false);
    }
}

5 SparkStreaming的运行流程

5.1　图解说明

5.2　文字解说

1、我们在集群中的其中一台机器上提交我们的Application Jar，然后就会产生一个Application，开启一个Driver，然后初始化SparkStreaming的程序入口StreamingContext；

2、Master会为这个Application的运行分配资源，在集群中的一台或者多台Worker上面开启Excuter，executer会向Driver注册；

3、Driver服务器会发送多个receiver给开启的excuter，（receiver是一个接收器，是用来接收消息的，在excuter里面运行的时候，其实就相当于一个task任务）

4、receiver接收到数据后，每隔200ms就生成一个block块，就是一个rdd的分区，然后这些block块就存储在executer里面，block块的存储级别是Memory_And_Disk_2；

5、receiver产生了这些block块后会把这些block块的信息发送给StreamingContext；

6、StreamingContext接收到这些数据后，会根据一定的规则将这些产生的block块定义成一个rdd；

6 Driver HA（Standalone或者Mesos）

因为SparkStreaming是7*24小时运行，Driver只是一个简单的进程，有可能挂掉，所以实现Driver的HA就有必要（如果使用的Client模式就无法实现Driver HA ，这里针对的是cluster模式）。Yarn平台cluster模式提交任务，AM(AplicationMaster)相当于Driver，如果挂掉会自动启动AM。这里所说的DriverHA针对的是Spark standalone和Mesos资源调度的情况下。实现Driver的高可用有两个步骤:

第一：提交任务层面，在提交任务的时候加上选项 --supervise,当Driver挂掉的时候会自动重启Driver。

第二：代码层面，使用JavaStreamingContext.getOrCreate（checkpoint路径，JavaStreamingContextFactory）

• Driver中元数据包括：

1. 创建应用程序的配置信息。
2. DStream的操作逻辑。
3. job中没有完成的批次数据，也就是job的执行进度。

优质内容筛选与推荐>>
1、新技术变革下保险业的挑战与机遇
2、安装SQL SERVER时提示"以前的某个程序安装已在安装计算机上创建挂起的文件操作 "解决方法
3、几种数据层的适用场景
4、运算符和typeof
5、BageCMS上传文件错误 fix