Flink DataStream Manager(托管) Keyed State的简单使用

女爷i 2021-12-18 14:45 285阅读 0赞

## Flink DataStream Manager(托管) Keyed State的简单使用 ##

Manager（托管）Keyed State接口提供对不同类型状态的访问，这些状态都限定为当前输入元素的键。这意味着这种类型的状态只能用于一个 `KeyedStream`，可以通过创建`stream.keyBy(…)`返回`KeyedStream`对象。

这个可用的State有：

*  `ValueState<T>`：保存一个状态值，可以通过`update()`更新State，通过`value()`方法获取当前状态值。
 *  `ListState<T>`：保存一个元素列表的状态，可以通过`add()、addAll()`方法添加到`iterable`，在使用`update()`更新状态值。通过`value()`取出`Iterable`对象，然后通过`get()`取出单个值。
 *  `ReducingState<T>`：类似于ListState，但是添加的元素，需要在`ReduceFunction`中聚合。
 *  `AggregatingState<IN,OUT>`：这将保留一个值，该值表示添加到状态的所有值的聚合。与`ReducingState`相反，聚合类型可能不同于添加到状态的元素类型。接口与for liststate相同，但使用add（in）添加的元素是使用指定的aggregateFunction聚合的。
 *  `MapState<UK, UV>`：保留一系列的映射表，你可以将键值对放入状态，并在所有当前存储的映射上检索Iterable。使用`put(uk,uv)`或`putall(Map<UK，UV>)`添加映射。可以使用`get(uk)`检索与用户密钥关联的值。映射、键和值的iterable视图可以分别使用`entries()`、`keys()`和`values()`进行检索。

所有的types都有一个`clear()`方法，用于清除状态。

每一个State对象，都需要创建一个`StateDescriptor`，用于保存State的名称，以及获取状态的句柄。StateDescription有：`ValueStateDescriptor`、`ListStateDescriptor`、`ReducingStateDescriptor`、`MapStateDescriptor`等。

State的访问状态，可以通过`getRuntimeContext()`调用`getState(descriptor)`方法获取。具体如下：

*  `ValueState<T> getState(ValueStateDescriptor<T>)`
 *  `ReducingState<T> getReducingState(ReducingStateDescriptor<T>)`
 *  `ListState<T> getListState(ListStateDescriptor<T>)`
 *  `AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)`
 *  `FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC>)`
 *  `MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)`

下面看一个简单示例：

public class ManagerKeyedState { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            env
                    .fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
                    .keyBy(0)
                    .flatMap(new RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() { 
    
                        private ValueState<Tuple2<Long, Long>> state;
    
                        @Override
                        public void open(Configuration parameters) throws Exception { 
                            super.open(parameters);
                            ValueStateDescriptor<Tuple2<Long, Long>> descriptor = new ValueStateDescriptor<Tuple2<Long, Long>>("", TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() { 
                            }), Tuple2.of(0L, 0L));
    
                            state = getRuntimeContext().getState(descriptor);
                        }
    
                        @Override
                        public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception { 
                            Tuple2<Long, Long> currentSum = state.value();
                            System.out.println("currentSum: " + currentSum + " value: " + value);
                            currentSum.f0 += 1;
                            currentSum.f1 += value.f1;
    
                            state.update(currentSum);
    
                            if (currentSum.f0 >= 2) { 
                                out.collect(Tuple2.of(value.f0, currentSum.f1 / currentSum.f0));
                                state.clear();
                            }
                        }
                    })
                    .print();
    
            env.execute("ManagerKeyedState");
        }
    }

输出结果：

currentSum: (0,0)  value: (1,3)
    currentSum: (1,3)  value: (1,5)
    (1,4)
    currentSum: (0,0)  value: (1,7)
    currentSum: (1,7)  value: (1,4)
    (1,5)
    currentSum: (0,0)  value: (1,2)