844      阿裏雲  技術社區[雲棲]

Flink - Working with State

All transformations in Flink may look like functions (in the functional processing terminology), but are in fact stateful operators. 
You can make every transformation (map, filter, etc) stateful by using Flink’s state interface or checkpointing instance fields of your function. 
You can register any instance field as managed state by implementing an interface. 
In this case, and also in the case of using Flink’s native state interface, Flink will automatically take consistent snapshots of your state periodically, and restore its value in the case of a failure.

討論如何使用Flink的state接口來管理狀態數據，對於這些狀態數據，Flink會自動的定期做snapshots，並且當failure後，會自動restore這些狀態

State主要可以分為兩種，Keyed State and Operator State

Keyed State

Keyed State is always relative to keys and can only be used in functions and operators on a KeyedStream.

You can think of Keyed State as Operator State that has been partitioned, or sharded, with exactly one state-partition per key. Each keyed-state is logically bound to a unique composite of <parallel-operator-instance, key>, and since each key “belongs” to exactly one parallel instance of a keyed operator, we can think of this simply as <operator, key>.

Keyed State is further organized into so-called Key Groups. Key Groups are the atomic unit by which Flink can redistribute Keyed State; there are exactly as many Key Groups as the defined maximum parallelism. During execution each parallel instance of a keyed operator works with the keys for one or more Key Groups.

Keyed state 隻能用於KeyedStream

Keyed state會以key做partitioned, or sharded，每個keyed-state都會關聯一個parallel-operator-instance

Keyed state的問題在於，在並發度增加時，需要把Keyed State切分開

為了便於keyed state的遷移和管理，實現Key Groups，這是Flink redistribute的最小單位

Operator State

With Operator State (or non-keyed state), each operator state is bound to one parallel operator instance. The Kafka source connector is a good motivating example for the use of Operator State in Flink. Each parallel instance of this Kafka consumer maintains a map of topic partitions and offsets as its Operator State.

The Operator State interfaces support redistributing state among parallel operator instances when the parallelism is changed. There can be different schemes for doing this redistribution; the following are currently defined:

• List-style redistribution: Each operator returns a List of state elements. The whole state is logically a concatenation of all lists. On restore/redistribution, the list is evenly divided into as many sublists as there are parallel operators. Each operator gets a sublist, which can be empty, or contain one or more elements.

Operator State就是non-keyed state，比如Kafka source connector，這種operator state在做redistribution，比較簡單

Raw and Managed State

Keyed State and Operator State exist in two forms: managed and raw.

Managed State is represented in data structures controlled by the Flink runtime, such as internal hash tables, or RocksDB. Examples are “ValueState”, “ListState”, etc. Flink’s runtime encodes the states and writes them into the checkpoints.

Raw State is state that operators keep in their own data structures. When checkpointed, they only write a sequence of bytes into the checkpoint. Flink knows nothing about the state’s data structures and sees only the raw bytes.

All datastream functions can use managed state, but the raw state interfaces can only be used when implementing operators. Using managed state (rather than raw state) is recommended, since with managed state Flink is able to automatically redistribute state when the parallelism is changed, and also do better memory management.

Raw是沒有被Flink管理的狀態

Using Managed Keyed State

The Key/Value state interface provides access to different types of state that are all scoped to the key of the current input element. 
This means that this type of state can only be used on a KeyedStream, which can be created via stream.keyBy(…).

Key/Value state 隻能用於KeyedStream

Now, we will first look at the different types of state available and then we will see how they can be used in a program. The available state primitives are:

• ValueState<T>: This keeps a value that can be updated and retrieved (scoped to key of the input element, mentioned above, so there will possibly be one value for each key that the operation sees). The value can be set using update(T)and retrieved using T value().

• ListState<T>: This keeps a list of elements. You can append elements and retrieve an Iterable over all currently stored elements. Elements are added using add(T), the Iterable can be retrieved using Iterable<T> get().

• ReducingState<T>: This keeps a single value that represents the aggregation of all values added to the state. The interface is the same as for ListState but elements added using add(T) are reduced to an aggregate using a specifiedReduceFunction.

All types of state also have a method clear() that clears the state for the currently active key (i.e. the key of the input element).

3種不同類型的state，

ValueState，單值的state，可以通過update(T)或T value()來操作

ListState<T>， 多隻的state，通過add(T)或Iterable<T> get()來操作和訪問

ReducingState<T>，多值狀態，但是隻保留reduce的結果

並且所有的state，都有clear，來清除狀態數據

It is important to keep in mind that these state objects are only used for interfacing with state. The state is not necessarily stored inside but might reside on disk or somewhere else. 
The second thing to keep in mind is that the value you get from the state depend on the key of the input element. 
So the value you get in one invocation of your user function can be different from the one you get in another invocation if the key of the element is different.

這些state對象隻能被狀態接口使用， 
並且取出的狀態對象，取決於input element的key；所以不同的調用user function 得到的state value是不一樣的，因為element的key 可能不同

To get a state handle you have to create a StateDescriptor this holds the name of the state (as we will later see you can create several states, and they have to have unique names so that you can reference them), the type of the values that the state holds and possibly a user-specified function, such as a ReduceFunction. Depending on what type of state you want to retrieve you create one of ValueStateDescriptor, ListStateDescriptor or ReducingStateDescriptor.

對於state，需要一個StateDescriptor ，作為name用於reference這個state，如果你定義多個state，他們的StateDescriptor 必須是unique的。 
不同類型的state，有不同類型的StateDescriptor

State is accessed using the RuntimeContext, so it is only possible in rich functions. 
Please see here for information about that but we will also see an example shortly. 
The RuntimeContext that is available in a RichFunction has these methods for accessing state:

• ValueState<T> getState(ValueStateDescriptor<T>)
• ReducingState<T> getReducingState(ReducingStateDescriptor<T>)
• ListState<T> getListState(ListStateDescriptor<T>)

State對象通過RuntimeContext的接口獲取到，當然不同類型的state，對應於不同的接口； 
關鍵是，如果要使用state，必須要使用rich function，用普通的function是無法獲取到的

This is an example FlatMapFunction that shows how all of the parts fit together:

public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {

    /**
     * The ValueState handle. The first field is the count, the second field a running sum.
     */
    private transient ValueState<Tuple2<Long, Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {

        // access the state value
        Tuple2<Long, Long> currentSum = sum.value();

        // update the count
        currentSum.f0 += 1;

        // add the second field of the input value
        currentSum.f1 += input.f1;

        // update the state
        sum.update(currentSum);

        // if the count reaches 2, emit the average and clear the state
        if (currentSum.f0 >= 2) {
            out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
            sum.clear();
        }
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
                        Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
        sum = getRuntimeContext().getState(descriptor);
    }
}

// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
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 CountWindowAverage())
        .print();

Using Managed Operator State

A stateful function can implement either the more general CheckpointedFunction interface, or the ListCheckpointed<T extends Serializable> interface.

In both cases, the non-keyed state is expected to be a List of serializable objects, independent from each other, thus eligible for redistribution upon rescaling. In other words, these objects are the finest granularity at which non-keyed state can be repartitioned. As an example, if with parallelism 1 the checkpointed state of the BufferingSink contains elements(test1, 2) and (test2, 2), when increasing the parallelism to 2, (test1, 2) may end up in task 0, while (test2, 2) will go to task 1.

Operator state，即non-keyed state , 被表示為serializable 對象列表，這些對象間是無關的，所以在變更parallelism 時，隻需要簡單的repartitioned

可以通過實現ListCheckpointed<T extends Serializable>或CheckpointedFunction接口，來實現對operator state的管理

ListCheckpointed

The ListCheckpointed interface requires the implementation of two methods:

List<T> snapshotState(long checkpointId, long timestamp) throws Exception;

void restoreState(List<T> state) throws Exception;

On snapshotState() the operator should return a list of objects to checkpoint and restoreState has to handle such a list upon recovery. 
If the state is not re-partitionable, you can always return a Collections.singletonList(MY_STATE) in thesnapshotState().

Collections.singletonList表示不可變列表

CheckpointedFunction

The CheckpointedFunction interface also requires the implementation of two methods:

void snapshotState(FunctionSnapshotContext context) throws Exception;

void initializeState(FunctionInitializationContext context) throws Exception;

Whenever a checkpoint has to be performed snapshotState() is called. The counterpart, initializeState(), is called every time the user-defined function is initialized, be that when the function is first initialized or be that when actually recovering from an earlier checkpoint. Given this, initializeState() is not only the place where different types of state are initialized, but also where state recovery logic is included.

This is an example of a function that uses CheckpointedFunction, a stateful SinkFunction that uses state to buffer elements before sending them to the outside world:

給個例子，stateful sinkfunction，在發送前先cache，

public class BufferingSink
        implements SinkFunction<Tuple2<String, Integer>>,
                   CheckpointedFunction,
                   CheckpointedRestoring<ArrayList<Tuple2<String, Integer>>> {

    private final int threshold;

    private transient ListState<Tuple2<String, Integer>> checkpointedState;

    private List<Tuple2<String, Integer>> bufferedElements;

    public BufferingSink(int threshold) {
        this.threshold = threshold;
        this.bufferedElements = new ArrayList<>();
    }

    @Override
    public void invoke(Tuple2<String, Integer> value) throws Exception {
        bufferedElements.add(value);
        if (bufferedElements.size() == threshold) {
            for (Tuple2<String, Integer> element: bufferedElements) {
                // send it to the sink
            }
            bufferedElements.clear();
        }
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        checkpointedState = context.getOperatorStateStore().
            getSerializableListState("buffered-elements"); //通過context初始化state

        if (context.isRestored()) { //如果context中有可以restore的數據
            for (Tuple2<String, Integer> element : checkpointedState.get()) { //restore
                bufferedElements.add(element);
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        checkpointedState.clear(); //清空
        for (Tuple2<String, Integer> element : bufferedElements) {
            checkpointedState.add(element); //snapshot
        }
    }

    @Override
    public void restoreState(ArrayList<Tuple2<String, Integer>> state) throws Exception { //這幹嘛用的？
        // this is from the CheckpointedRestoring interface.
        this.bufferedElements.addAll(state);
    }
}

Stateful Source Functions

Stateful sources require a bit more care as opposed to other operators. 
In order to make the updates to the state and output collection atomic (required for exactly-once semantics on failure/recovery), the user is required to get a lock from the source’s context.

對於有狀態的source，有些不一樣的是，在更新state和output時，注意要加鎖來保證exactly-once，比如避免多個線程同時更新offset

public static class CounterSource
        extends RichParallelSourceFunction<Long>
        implements ListCheckpointed<Long> {

    /**  current offset for exactly once semantics */
    private Long offset;

    /** flag for job cancellation */
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<Long> ctx) {
        final Object lock = ctx.getCheckpointLock();

        while (isRunning) {
            // output and state update are atomic
            synchronized (lock) { //加鎖保證原子性
                ctx.collect(offset);
                offset += 1;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public List<Long> snapshotState(long checkpointId, long checkpointTimestamp) {
        return Collections.singletonList(offset); //不可變list，表示不可re-partitionable
    }

    @Override
    public void restoreState(List<Long> state) {
        for (Long s : state)
            offset = s;
    }
}

https://ci.apache.org/projects/flink/flink-docs-release-1.2/ops/state_backends.html

State Backends

Programs written in the Data Stream API often hold state in various forms:

• Windows gather elements or aggregates until they are triggered
• Transformation functions may use the key/value state interface to store values
• Transformation functions may implement the Checkpointed interface to make their local variables fault tolerant

主要的state，包含幾種，

windows裏麵gather的elements 
Transformation functions中用key/value state interface創建的state 
Transformation functions 中通過Checkpointed interface 去對local variables做的state

When checkpointing is activated, such state is persisted upon checkpoints to guard against data loss and recover consistently. 
How the state is represented internally, and how and where it is persisted upon checkpoints depends on the chosen State Backend.

關鍵，state如何和存到何處，還是看具體用什麼State Backend

Available State Backends

Out of the box, Flink bundles these state backends:

• MemoryStateBacked
• FsStateBackend
• RocksDBStateBackend

If nothing else is configured, the system will use the MemoryStateBacked.

The MemoryStateBackend

The MemoryStateBacked holds data internally as objects on the Java heap. Key/value state and window operators hold hash tables that store the values, triggers, etc.

Upon checkpoints, this state backend will snapshot the state and send it as part of the checkpoint acknowledgement messages to the JobManager (master), which stores it on its heap as well.

MemoryStateBackend顧名思義，就是state是存儲在Java heap中的；在做checkpoints的時候，state backend 會將state snapshot放入 checkpoint acknowledgement messages 發給JobManager，JobManager 仍然是將它存在heap中。

The FsStateBackend

The FsStateBackend is configured with a file system URL (type, address, path), such as for example “hdfs://namenode:40010/flink/checkpoints” or “file:///data/flink/checkpoints”.

The FsStateBackend holds in-flight data in the TaskManager’s memory. Upon checkpointing, it writes state snapshots into files in the configured file system and directory. 
Minimal metadata is stored in the JobManager’s memory (or, in high-availability mode, in the metadata checkpoint).

State snapshot數據是存在文件係統中的，而JobManager的內存中，隻是存放最小的元數據

The RocksDBStateBackend

隻是用RocksDB來替換文件係統，

NOTE: To use the RocksDBStateBackend you also have to add the correct maven dependency to your project:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-statebackend-rocksdb_2.10</artifactId>
  <version>1.0.3</version>
</dependency>

The backend is currently not part of the binary distribution. See here for an explanation of how to include it for cluster execution.

Configuring a State Backend

State backends can be configured per job. In addition, you can define a default state backend to be used when the job does not explicitly define a state backend.

Setting the Per-job State Backend

The per-job state backend is set on the StreamExecutionEnvironment of the job, as shown in the example below:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));

Setting Default State Backend

A default state backend can be configured in the flink-conf.yaml, using the configuration key state.backend.

Possible values for the config entry are jobmanager (MemoryStateBackend), filesystem (FsStateBackend), or the fully qualified class name of the class that implements the state backend factory FsStateBackendFactory.

In the case where the default state backend is set to filesystem, the entry state.backend.fs.checkpointdir defines the directory where the checkpoint data will be stored.

A sample section in the configuration file could look as follows:

# The backend that will be used to store operator state checkpoints

state.backend: filesystem


# Directory for storing checkpoints

state.backend.fs.checkpointdir: hdfs://namenode:40010/flink/checkpoints

最後更新：2017-04-07 21:05:50

  上一篇： kudu

  下一篇： Kafka 0.10.0