一、ListState的方法
- get()方法获取值
- add(IN value)，addAll(List values)方法更新值
- update(List values) 用新List 替换 原来的List
- clear() 清空List，List还存在，但是没有元素
二、代码案例
1. 需求
记录每个人的历史操作信息，例如输入
1,login
1,buy something
1,logout
2,login
2,click book
1,login
要输出
(1,[login])
(1,[login, buy something])
(1,[login, buy something, logout])
(2,[login])
(2,[login, click book])
(1,[login, buy something, logout, login])
2. 主体
public class Test03_ListState {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<String> source = env.socketTextStream(BaseConstant.URL, BaseConstant.PORT);
SingleOutputStreamOperator<Tuple2<String, String>> dataStream = source
.map(new MapFunction<String, Tuple2<String, String>>() {
@Override
public Tuple2<String, String> map(String value) throws Exception {
String[] split = value.split(",");
if (split != null && split.length == 2) {
return Tuple2.of(split[0], split[1]);
}
return null;
}
});
dataStream
.keyBy(t -> t.f0)
.process(new MyKeyedProcessFunction())
.print();
env.execute();
}
}
3. 处理方法
public static class MyKeyedProcessFunction extends KeyedProcessFunction<String, Tuple2<String, String>, Tuple2<String, List<String>>> {
//之前的操作记录
private transient ListState<String> listState;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
ListStateDescriptor<String> recentOperatorsDescriptor = new ListStateDescriptor<String>(
"recent-operator",
String.class);
listState = getRuntimeContext().getListState(recentOperatorsDescriptor);
}
@Override
public void processElement(Tuple2<String, String> value, Context ctx, Collector<Tuple2<String, List<String>>> out) throws Exception {
String action = value.f1;
listState.add(action);
Iterable<String> iterable = listState.get();
ArrayList<String> events = new ArrayList<>();
for (String actionName : iterable) {
events.add(actionName);
}
out.collect(Tuple2.of(value.f0, events));
listState.update(events);
}
}
4. 运行结果
5. 用ValueState实现ListState
用ValueState实现ListState可以实现ListState，只要ValueState里装的是List
public static class MyKeyedProcessFunction extends KeyedProcessFunction<String, Tuple2<String, String>, Tuple2<String, List<String>>> {
//之前的操作记录
private transient ValueState<List<String>> listState;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
ValueStateDescriptor<List<String>> recentOperatorsDescriptor = new ValueStateDescriptor<List<String>>(
"recent-operator",
TypeInformation.of(new TypeHint<List<String>>(){
}));
listState = getRuntimeContext().getState(recentOperatorsDescriptor);
}
@Override
public void processElement(Tuple2<String, String> value, Context ctx, Collector<Tuple2<String, List<String>>> out) throws Exception {
String action = value.f1;
List<String> lst = listState.value();
if (lst == null){
lst = new ArrayList<>();
}
lst.add(action);
listState.update(lst);
out.collect(Tuple2.of(value.f0, listState.value()));
}
}