Spark 检查点

小咪咪 2022-05-12 03:42 187阅读 0赞

1、设置检查点的代码基本如下，供参考.

public static void main(String[] args) throws Exception {
            Function0<JavaStreamingContext> contextFactory = new Function0<JavaStreamingContext>() {
                public JavaStreamingContext call() {
                    SparkConf conf = new SparkConf();
                    conf.setMaster("local[4]");
                    conf.setAppName("wc");
                    JavaStreamingContext jssc = new JavaStreamingContext(conf,new Duration(2000));
                    JavaDStream<String> lines = jssc.socketTextStream("192.168.44.128",9999);
                    JavaDStream<Long> dsCount = lines.countByWindow(new Duration(24 * 60 * 60 * 1000),new Duration(2000));
                    dsCount.print();
                    jssc.checkpoint("file:///d:/scala/check");
                    return jssc;
                }
            };
    
            JavaStreamingContext context = JavaStreamingContext.getOrCreate("file:///d:/scala/check", contextFactory);
    
            context.start();
            context.awaitTermination();
        }

2、UpdateStateByKey

一般是针对键值对的操作.

public static void main(String[] args) throws Exception {
            SparkConf conf = new SparkConf();
            conf.setAppName("wc");
            conf.setMaster("local[4]");
            //创建Spark流应用上下文
            JavaStreamingContext jsc = new JavaStreamingContext(conf, Seconds.apply(3));
    
            jsc.checkpoint("file:///d:/scala/check");
            //创建socket离散流
            JavaReceiverInputDStream sock = jsc.socketTextStream("192.1687.44.128",9999);
            //压扁
            JavaDStream<String> wordsDS = sock.flatMap(new FlatMapFunction<String,String>() {
                public Iterator call(String str) throws Exception {
                    List<String> list = new ArrayList<String>() ;
                    String[] arr = str.split(" ");
                    for(String s : arr){
                        list.add(s);
                    }
                    return list.iterator();
                }
            });
    
            //映射成元组
            JavaPairDStream<String,Integer> pairDS = wordsDS.mapToPair(new PairFunction<String, String, Integer>() {
                public Tuple2<String, Integer> call(String s) throws Exception {
                    return new Tuple2<String,Integer>(s,1);
                }
            }) ;
    
            JavaPairDStream<String,Integer> jps = pairDS.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
                public Optional<Integer> call(List<Integer> v1, Optional<Integer> v2) throws Exception {
                    Integer newCount = v2.isPresent() ? v2.get() : 0  ;
    
                    System.out.println("old value : " + newCount);
                    for(Integer i : v1){
                        System.out.println("new value : " + i);
                        newCount = newCount +  i;
                    }
                    return Optional.of(newCount);
                }
            });
    
    
    
            //聚合
            JavaPairDStream<String,Integer> countDS = jps.reduceByKey(new Function2<Integer, Integer, Integer>() {
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1 + v2;
                }
            });
    
    
            //打印
            countDS.print();
    
            jsc.start();
    
            jsc.awaitTermination();
    
            jsc.stop();
        }