首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >SparkStream mapWithState编程练习

SparkStream mapWithState编程练习

作者头像
birdskyws
发布2018-09-12 15:02:34
8590
发布2018-09-12 15:02:34
举报

SparkStream在处理流数据时,按时间间隔把数据分成小批,在一个小批中利用RDD 的函数完成各种运算。如果要在各小批之间共享数据,或者保存到每批次的数据到一个集中变量中,就要用到mapWithState函数,在整个流计算任务中维护了一个key-value State对象(应该也是一个RDD),根据本批次的任务更改State。本文是mapWithState的实例代码。

完整代码

    public static void main(String[] args) throws InterruptedException {
        String checkpointDirectory = "/Users/wangsen/hadoop/checkdir/check4";
        // Create the context with a 1 second batch size
        SparkConf sparkConf = new SparkConf().setAppName("JavaStatefulNetworkWordCount");
        JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(5));
        ssc.checkpoint(checkpointDirectory);
        ssc.sparkContext().setLogLevel("ERROR");
        JavaReceiverInputDStream<String> lines = ssc.socketTextStream(
                "localhost", 9999, StorageLevels.MEMORY_AND_DISK_SER_2);

        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(SPACE.split(x)).iterator());

        JavaPairDStream<String, Integer> wordsDstream = words.mapToPair(s -> new Tuple2<>(s, 1));

        // Update the cumulative count function
        // 注释三
        Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>> mappingFunc =
                (word, one, state) -> {
                    int sum = one.orElse(0) + (state.exists() ? state.get() : 0);
                    Tuple2<String, Integer> output = new Tuple2<>(word, sum);
                    state.update(sum);
                    return output;
                };
        // DStream made of get cumulative counts that get updated in every batch
        // 注释一:
        JavaMapWithStateDStream<String, Integer, Integer, Tuple2<String, Integer>> stateDstream =
                wordsDstream.mapWithState(StateSpec.function(mappingFunc));
        //注释二:
        stateDstream.print();
        stateDstream.stateSnapshots().print();
        ssc.start();
        ssc.awaitTermination();
    }

实验

nc -lk 9999 a b c d e f

输出,第一个输出stateDStream输出,第二个stateDStream.stateSnapshots输出

-------------------------------------------
Time: 1535529530000 ms
-------------------------------------------
(A,1)
(B,1)
(C,1)

-------------------------------------------
Time: 1535529530000 ms
-------------------------------------------
(A,1)
(B,1)
(C,1)

A B e f

-------------------------------------------
Time: 1535529630000 ms
-------------------------------------------
(e,1)
(A,2)
(B,2)
(f,1)

-------------------------------------------
Time: 1535529630000 ms
-------------------------------------------
(e,1)
(A,2)
(B,2)
(f,1)
(C,1)

stateDStream只做增量,获取全局使用stateSnapshots。

代码注释

1、JavaMapWithStateDStream 对象(注释一)。

统计词频生成JavaPairDStream对象wordsCount,wordCount调用mapWithState生成JavaMapWithStateDStream对象。

2、打印JavaMapWithStateDStream对象(注释二)。

state.print(); # 只打印本轮任务。 state.stateSnapshots().print(); # 获取全部state的值,并打印

3、生成、更新state对象(注释三)。

Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>> mappingFunc Function3参数说明 (1)String输入值,代表要更新的State对象Key, (2)Optional<Integer>输入值,代表本批次计算得到key对应的value值, (3)State<Integer>输入值,有点类似回调值,在State中保存的value值,旧的值,调用函数的时候已经赋值。在代码里可以实现创建更新等操作:可以累加;可以比较大小,更新一个更大值,等等。 (4)Tuple2<String, Integer>返回值,State的一个item。返回Tuple2就更新State中相应Key的数据,调用remove可以删除State中的Key对象。 Tuple2<String,Integer>定义了State类型。

总结

mapWithState能将batch的数据统一保存,功能很强大,就是理解函数参数有些费劲。本文通过代码介绍了该api使用方法,以后需要的时候能快速copy代码。

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2018.08.29 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 完整代码
  • 实验
  • 代码注释
    • 1、JavaMapWithStateDStream 对象(注释一)。
      • 2、打印JavaMapWithStateDStream对象(注释二)。
        • 3、生成、更新state对象(注释三)。
        • 总结
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档