Package org.apache.flink.streaming.runtime.operators.windowing

Class WindowOperator.MergingWindowStateStore

    • Method Detail

      • getState

        public <T> ValueState<T> getState​(ValueStateDescriptor<T> stateProperties)
        Description copied from interface: KeyedStateStore
        Gets a handle to the system's key/value state. The key/value state is only accessible if the function is executed on a KeyedStream. On each access, the state exposes the value for the key of the element currently processed by the function. Each function may have multiple partitioned states, addressed with different names.

        Because the scope of each value is the key of the currently processed element, and the elements are distributed by the Flink runtime, the system can transparently scale out and redistribute the state and KeyedStream.

        The following code example shows how to implement a continuous counter that counts how many times elements of a certain key occur, and emits an updated count for that element on each occurrence. 

        
 DataStream<MyType> stream = ...;
 KeyedStream<MyType> keyedStream = stream.keyBy("id");

 keyedStream.map(new RichMapFunction<MyType, Tuple2<MyType, Long>>() {

     private ValueState<Long> state;

     public void open(OpenContext ctx) {
         state = getRuntimeContext().getState(
                 new ValueStateDescriptor<Long>("count", LongSerializer.INSTANCE, 0L));
     }

     public Tuple2<MyType, Long> map(MyType value) {
         long count = state.value() + 1;
         state.update(count);
         return new Tuple2<>(value, count);
     }
 });
        Specified by:
        getState in interface KeyedStateStore
        Overrides:
        getState in class DefaultKeyedStateStore
        Type Parameters:
        T - The type of value stored in the state.
        Parameters:
        stateProperties - The descriptor defining the properties of the stats.
        Returns:
        The partitioned state object.

      • getListState

        public <T> ListState<T> getListState​(ListStateDescriptor<T> stateProperties)
        Description copied from interface: KeyedStateStore
        Gets a handle to the system's key/value list state. This state is similar to the state accessed via KeyedStateStore.getState(ValueStateDescriptor), but is optimized for state that holds lists. One can adds elements to the list, or retrieve the list as a whole.

        This state is only accessible if the function is executed on a KeyedStream. 

        
 DataStream<MyType> stream = ...;
 KeyedStream<MyType> keyedStream = stream.keyBy("id");

 keyedStream.map(new RichFlatMapFunction<MyType, List<MyType>>() {

     private ListState<MyType> state;

     public void open(OpenContext ctx) {
         state = getRuntimeContext().getListState(
                 new ListStateDescriptor<>("myState", MyType.class));
     }

     public void flatMap(MyType value, Collector<MyType> out) {
         if (value.isDivider()) {
             for (MyType t : state.get()) {
                 out.collect(t);
             }
         } else {
             state.add(value);
         }
     }
 });
        Specified by:
        getListState in interface KeyedStateStore
        Overrides:
        getListState in class DefaultKeyedStateStore
        Type Parameters:
        T - The type of value stored in the state.
        Parameters:
        stateProperties - The descriptor defining the properties of the stats.
        Returns:
        The partitioned state object.

      • getReducingState

        public <T> ReducingState<T> getReducingState​(ReducingStateDescriptor<T> stateProperties)
        Description copied from interface: KeyedStateStore
        Gets a handle to the system's key/value reducing state. This state is similar to the state accessed via KeyedStateStore.getState(ValueStateDescriptor), but is optimized for state that aggregates values.

        This state is only accessible if the function is executed on a KeyedStream. 

        
 DataStream<MyType> stream = ...;
 KeyedStream<MyType> keyedStream = stream.keyBy("id");

 keyedStream.map(new RichMapFunction<MyType, List<MyType>>() {

     private ReducingState<Long> state;

     public void open(OpenContext ctx) {
         state = getRuntimeContext().getReducingState(
                 new ReducingStateDescriptor<>("sum", (a, b) -> a + b, Long.class));
     }

     public Tuple2<MyType, Long> map(MyType value) {
         state.add(value.count());
         return new Tuple2<>(value, state.get());
     }
 });
        Specified by:
        getReducingState in interface KeyedStateStore
        Overrides:
        getReducingState in class DefaultKeyedStateStore
        Type Parameters:
        T - The type of value stored in the state.
        Parameters:
        stateProperties - The descriptor defining the properties of the stats.
        Returns:
        The partitioned state object.

      • getAggregatingState

        public <IN,​ACC,​OUT> AggregatingState<IN,​OUT> getAggregatingState​(AggregatingStateDescriptor<IN,​ACC,​OUT> stateProperties)
        Description copied from interface: KeyedStateStore
        Gets a handle to the system's key/value folding state. This state is similar to the state accessed via KeyedStateStore.getState(ValueStateDescriptor), but is optimized for state that aggregates values with different types.

        This state is only accessible if the function is executed on a KeyedStream. 

        
 DataStream<MyType> stream = ...;
 KeyedStream<MyType> keyedStream = stream.keyBy("id");
 AggregateFunction<...> aggregateFunction = ...

 keyedStream.map(new RichMapFunction<MyType, List<MyType>>() {

     private AggregatingState<MyType, Long> state;

     public void open(OpenContext ctx) {
         state = getRuntimeContext().getAggregatingState(
                 new AggregatingStateDescriptor<>("sum", aggregateFunction, Long.class));
     }

     public Tuple2<MyType, Long> map(MyType value) {
         state.add(value);
         return new Tuple2<>(value, state.get());
     }
 });
        Specified by:
        getAggregatingState in interface KeyedStateStore
        Overrides:
        getAggregatingState in class DefaultKeyedStateStore
        Type Parameters:
        IN - The type of the values that are added to the state.
        ACC - The type of the accumulator (intermediate aggregation state).
        OUT - The type of the values that are returned from the state.
        Parameters:
        stateProperties - The descriptor defining the properties of the stats.
        Returns:
        The partitioned state object.

      • getMapState

        public <UK,​UV> MapState<UK,​UV> getMapState​(MapStateDescriptor<UK,​UV> stateProperties)
        Description copied from interface: KeyedStateStore
        Gets a handle to the system's key/value map state. This state is similar to the state accessed via KeyedStateStore.getState(ValueStateDescriptor), but is optimized for state that is composed of user-defined key-value pairs

        This state is only accessible if the function is executed on a KeyedStream. 

        
 DataStream<MyType> stream = ...;
 KeyedStream<MyType> keyedStream = stream.keyBy("id");

 keyedStream.map(new RichMapFunction<MyType, List<MyType>>() {

     private MapState<MyType, Long> state;

     public void open(OpenContext ctx) {
         state = getRuntimeContext().getMapState(
                 new MapStateDescriptor<>("sum", MyType.class, Long.class));
     }

     public Tuple2<MyType, Long> map(MyType value) {
         return new Tuple2<>(value, state.get(value));
     }
 });
        Specified by:
        getMapState in interface KeyedStateStore
        Overrides:
        getMapState in class DefaultKeyedStateStore
        Type Parameters:
        UK - The type of the user keys stored in the state.
        UV - The type of the user values stored in the state.
        Parameters:
        stateProperties - The descriptor defining the properties of the stats.
        Returns:
        The partitioned state object.