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
