SparkStream|SparkStream mapWithState编程练习 SparkStreammapWithState编程练习

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 lines = ssc.socketTextStream( "localhost", 9999, StorageLevels.MEMORY_AND_DISK_SER_2); JavaDStream words = lines.flatMap(x -> Arrays.asList(SPACE.split(x)).iterator()); JavaPairDStream wordsDstream = words.mapToPair(s -> new Tuple2<>(s, 1)); // Update the cumulative count function // 注释三 Function3, State, Tuple2> mappingFunc = (word, one, state) -> { int sum = one.orElse(0) + (state.exists() ? state.get() : 0); Tuple2 output = new Tuple2<>(word, sum); state.update(sum); return output; }; // DStream made of get cumulative counts that get updated in every batch // 注释一： JavaMapWithStateDStream> 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, State, Tuple2> mappingFunc
Function3参数说明
（1）String输入值，代表要更新的State对象Key,
（2）Optional输入值，代表本批次计算得到key对应的value值，
（3）State输入值，有点类似回调值，在State中保存的value值，旧的值，调用函数的时候已经赋值。在代码里可以实现创建更新等操作：可以累加；可以比较大小，更新一个更大值，等等。
（4）Tuple2返回值，State的一个item。返回Tuple2就更新State中相应Key的数据，调用remove可以删除State中的Key对象。
Tuple2定义了State类型。
总结 【SparkStream|SparkStream mapWithState编程练习】mapWithState能将batch的数据统一保存，功能很强大，就是理解函数参数有些费劲。本文通过代码介绍了该api使用方法，以后需要的时候能快速copy代码。