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代码。

我的博客即将搬运同步至腾讯云+社区，邀请大家一同入驻：https://cloud.tencent.com/developer/support-plan?invite_code=1plqfa9at9d1z