序

本文主要研究一下flink的Managed Keyed State

State

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/State.java

/**
 * Interface that different types of partitioned state must implement.
 *
 * <p>The state is only accessible by functions applied on a {@code KeyedStream}. The key is
 * automatically supplied by the system, so the function always sees the value mapped to the
 * key of the current element. That way, the system can handle stream and state partitioning
 * consistently together.
 */
@PublicEvolving
public interface State {

    /**
     * Removes the value mapped under the current key.
     */
    void clear();
}

• State是所有不同類型的State必須實(shí)現(xiàn)的接口胸完，它定義了clear方法

ValueState

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/ValueState.java

@PublicEvolving
public interface ValueState<T> extends State {

    /**
     * Returns the current value for the state. When the state is not
     * partitioned the returned value is the same for all inputs in a given
     * operator instance. If state partitioning is applied, the value returned
     * depends on the current operator input, as the operator maintains an
     * independent state for each partition.
     *
     * <p>If you didn't specify a default value when creating the {@link ValueStateDescriptor}
     * this will return {@code null} when to value was previously set using {@link #update(Object)}.
     *
     * @return The state value corresponding to the current input.
     *
     * @throws IOException Thrown if the system cannot access the state.
     */
    T value() throws IOException;

    /**
     * Updates the operator state accessible by {@link #value()} to the given
     * value. The next time {@link #value()} is called (for the same state
     * partition) the returned state will represent the updated value. When a
     * partitioned state is updated with null, the state for the current key
     * will be removed and the default value is returned on the next access.
     *
     * @param value The new value for the state.
     *
     * @throws IOException Thrown if the system cannot access the state.
     */
    void update(T value) throws IOException;

}

• ValueState繼承了State接口赛不，它定義了value鸥滨、update兩個(gè)方法，一個(gè)用于取值品追，一個(gè)用于更新值

AppendingState

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/AppendingState.java

@PublicEvolving
public interface AppendingState<IN, OUT> extends State {

    /**
     * Returns the current value for the state. When the state is not
     * partitioned the returned value is the same for all inputs in a given
     * operator instance. If state partitioning is applied, the value returned
     * depends on the current operator input, as the operator maintains an
     * independent state for each partition.
     *
     * <p><b>NOTE TO IMPLEMENTERS:</b> if the state is empty, then this method
     * should return {@code null}.
     *
     * @return The operator state value corresponding to the current input or {@code null}
     * if the state is empty.
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    OUT get() throws Exception;

    /**
     * Updates the operator state accessible by {@link #get()} by adding the given value
     * to the list of values. The next time {@link #get()} is called (for the same state
     * partition) the returned state will represent the updated list.
     *
     * <p>If null is passed in, the state value will remain unchanged.
     *
     * @param value The new value for the state.
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    void add(IN value) throws Exception;

}

• AppendingState繼承了State接口，它定義了get、add方法液荸，該State接收IN、OUT兩個(gè)泛型

FoldingState

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/FoldingState.java

@PublicEvolving
@Deprecated
public interface FoldingState<T, ACC> extends AppendingState<T, ACC> {}

• FoldingState繼承了AppendingState脱篙，其中OUT泛型表示ACC娇钱，即累積值；FoldingState在Flink 1.4版本被標(biāo)記為廢棄绊困，后續(xù)會(huì)被移除掉文搂，可使用AggregatingState替代

MergingState

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/MergingState.java

/**
 * Extension of {@link AppendingState} that allows merging of state. That is, two instances
 * of {@link MergingState} can be combined into a single instance that contains all the
 * information of the two merged states.
 *
 * @param <IN> Type of the value that can be added to the state.
 * @param <OUT> Type of the value that can be retrieved from the state.
 */
@PublicEvolving
public interface MergingState<IN, OUT> extends AppendingState<IN, OUT> { }

• MergingState繼承了AppendingState，這里用命名表達(dá)merge state的意思秤朗，它有幾個(gè)子接口煤蹭，分別是ListState、ReducingState取视、AggregatingState

ListState

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/ListState.java

@PublicEvolving
public interface ListState<T> extends MergingState<T, Iterable<T>> {

    /**
     * Updates the operator state accessible by {@link #get()} by updating existing values to
     * to the given list of values. The next time {@link #get()} is called (for the same state
     * partition) the returned state will represent the updated list.
     *
     * <p>If null or an empty list is passed in, the state value will be null.
     *
     * @param values The new values for the state.
     *
     * @throws Exception The method may forward exception thrown internally (by I/O or functions).
     */
    void update(List<T> values) throws Exception;

    /**
     * Updates the operator state accessible by {@link #get()} by adding the given values
     * to existing list of values. The next time {@link #get()} is called (for the same state
     * partition) the returned state will represent the updated list.
     *
     * <p>If null or an empty list is passed in, the state value remains unchanged.
     *
     * @param values The new values to be added to the state.
     *
     * @throws Exception The method may forward exception thrown internally (by I/O or functions).
     */
    void addAll(List<T> values) throws Exception;
}

• ListState繼承了MergingState硝皂，它的OUT類型為Iterable<IN>；它主要用于operation存儲(chǔ)partitioned list state作谭，它繼承了MergingState接口(指定OUT的泛型為Iterable<T>)稽物，同時(shí)聲明了兩個(gè)方法；其中update用于全量更新state丢早，如果參數(shù)為null或者empty姨裸，那么state會(huì)被清空；addAll方法用于增量更新怨酝，如果參數(shù)為null或者empty傀缩，則保持不變，否則則新增給定的values

ReducingState

flink-core/1.7.0/flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/ReducingState.java

@PublicEvolving
public interface ReducingState<T> extends MergingState<T, T> {}

• ReducingState繼承了MergingState农猬，它的IN赡艰、OUT類型相同

AggregatingState

flink-core/1.7.0/flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/AggregatingState.java

@PublicEvolving
public interface AggregatingState<IN, OUT> extends MergingState<IN, OUT> {}

• AggregatingState繼承了MergingState，它與ReducingState不同斤葱，IN慷垮、OUT類型可以不同

MapState

flink-core-1.7.0-sources.jar!/org/apache/flink/api/common/state/MapState.java

@PublicEvolving
public interface MapState<UK, UV> extends State {

    /**
     * Returns the current value associated with the given key.
     *
     * @param key The key of the mapping
     * @return The value of the mapping with the given key
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    UV get(UK key) throws Exception;

    /**
     * Associates a new value with the given key.
     *
     * @param key The key of the mapping
     * @param value The new value of the mapping
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    void put(UK key, UV value) throws Exception;

    /**
     * Copies all of the mappings from the given map into the state.
     *
     * @param map The mappings to be stored in this state
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    void putAll(Map<UK, UV> map) throws Exception;

    /**
     * Deletes the mapping of the given key.
     *
     * @param key The key of the mapping
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    void remove(UK key) throws Exception;

    /**
     * Returns whether there exists the given mapping.
     *
     * @param key The key of the mapping
     * @return True if there exists a mapping whose key equals to the given key
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    boolean contains(UK key) throws Exception;

    /**
     * Returns all the mappings in the state.
     *
     * @return An iterable view of all the key-value pairs in the state.
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    Iterable<Map.Entry<UK, UV>> entries() throws Exception;

    /**
     * Returns all the keys in the state.
     *
     * @return An iterable view of all the keys in the state.
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    Iterable<UK> keys() throws Exception;

    /**
     * Returns all the values in the state.
     *
     * @return An iterable view of all the values in the state.
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    Iterable<UV> values() throws Exception;

    /**
     * Iterates over all the mappings in the state.
     *
     * @return An iterator over all the mappings in the state
     *
     * @throws Exception Thrown if the system cannot access the state.
     */
    Iterator<Map.Entry<UK, UV>> iterator() throws Exception;
}

• MapState直接繼承了State，它接收UK揍堕、UV兩個(gè)泛型料身，分別是map的key和value的類型

小結(jié)

• flink提供了好幾個(gè)不同類型的Managed Keyed State，有ValueState<T>衩茸、ListState<T>芹血、ReducingState<T>、AggregatingState<IN, OUT>、FoldingState<T, ACC>幔烛、MapState<UK, UV>
• ValueState<T>和MapState<UK, UV>是直接繼承State接口啃擦；FoldingState繼承了AppendingState<IN, OUT>(AppendingState直接繼承了State)；ListState饿悬、ReducingState令蛉、AggregatingState繼承了MergingState<IN, OUT>(MergingState繼承了AppendingState)
• FoldingState在Flink 1.4版本被標(biāo)記為廢棄，后續(xù)會(huì)被移除掉狡恬，可使用AggregatingState替代

doc

• Using Managed Keyed State