[大数据]Flink‘s State & Fault Tolerance

1. Working with State

本部分将介绍Flink提供的写状态程序的API。

1.1 Keyed DataStream

如果想使用键控状态（keyed state）,首先需要指定一个DataStream的key，以key将状态分区。通常使用keyBy(KeySelector/键选择器)指定DataStream的key，形成一个键控流（KeyedStream），允许在其上运用键控状态。
键选择器函数（key selector function）将处理的每条记录数据作为输入，返回一每条记录的key。key可以是任意的类型，但是必须从确定性的计算得出。
Flink的数据模型不是基于key-value对，因此不需要物理地将数据转换成k-v对，key是虚拟的。
下面是一个简单的例子展示键选择器函数返回对象的字段作为key:

// some ordinary POJO
public class WC {
  public String word;
  public int count;

  public String getWord() { return word; }
}
DataStream<WC> words = // [...]
KeyedStream<WC> keyed = words
  .keyBy(WC::getWord);

1.1.1 Tuple Keys and Expression Keys

有两种方法定义key，Tuple Keys 和 Expression Keys，可以分别用元组的索引或对象的字段指定key。使用 Expression Keys结合Java的lambda表达式非常简单且可以减少运行时开销。

2. Using Keyed State

有以下五种键空状态：

1. ValueState<T>

对每个key保存一个值，可以更新和取回;设置或更新使用update(T)方法，取回用T value()方法；

2. ListState<T>

对每个key保存其value成一个list，可以向其中添加元素和获取一个Tterable。添加元素用add(T)或addAll(List<T>)，获取元素可以使用Iterable<T> get()，也可以用update(List<T>)替换原来的list。

3. ReducingState<T>

保存被添加进来的值的聚合结果，和ListState相似，除了用add(T)添加进来的值被用一个ReduceFunction函数聚合。

4. AggregatingState<IN, OUT>

和ReducingState很相似，相比于ReduceState，它的聚合类型可能和输入值的类型不同；和ListState在添加元素时也很相似，除了被添加的元素被指定的AggregateFunction函数聚合。

5. MapState<UK, UV>

保存映射关系。可以存放key-value对和获取一个保存所有映射关系的Iterable。可以通过put(UK, UV)或者putAll(Map<UK, UV>)添加元素。可以通过get(UK)获取单个键所对应的值。获取所有的映射、所有的key、所有的value可以分别使用entries(), keys(), values()。也可以使用isEmpty()检查map是否有映射。

所有状态都有一个clear()方法用于清空当前key的状态。

创建状态，需要首先需要创建StateDescriptor。它保存状态的名称、状态值类型或者用户自定义的一个函数，如RecduceFunction。根据需要类型的状态，对应以下五种StateDescriptor:

ValueStateDescriptor
ListStateDescriptor
AggregatingStateDescriptor
ReducingStateDescriptor
MapStateDescriptor

使用状态需要运行时环境（RuntimeContext），只能来源于rich Functions。从RichFunction的运行是环境中可以通过以下方法获得所需要的状态：

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

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class ValueStateRFlatMap {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Tuple2<Long, Long>> inputStream = env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L));
        // 此处用到Java 的lambda表达式和Scala推演符号不同
        KeyedStream<Tuple2<Long, Long>, Long> tuple2LongKeyedStream = inputStream.keyBy(value -> value.f0);

        SingleOutputStreamOperator<Tuple2<Long, Long>> tuple2SingleOutput = tuple2LongKeyedStream.flatMap(new CountWindowAverage());
        // 打印结果
        tuple2SingleOutput.print();

        env.execute();
    }
}

class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
    // 定义值类型的状态 是一个两个值的Tuple类型
    private transient ValueState<Tuple2<Long, Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
        // 获取状态
        Tuple2<Long, Long> currentValue = sum.value();
        // 如果状态为空，则将值保存在状态中，否者输出(input.f0, (currentValue.f1 + input.f1) / 2))元组
        if (currentValue == null) {
            sum.update(input);
        } else {
            out.collect(Tuple2.of(input.f0, (currentValue.f1 + input.f1) / 2));
            sum.clear();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
//        sum = getRuntimeContext().getState(new ValueStateDescriptor<>("average"
//                , TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
//        })));

        // 用于设置状态存活周期
        StateTtlConfig stateTtlConfig = StateTtlConfig
                .newBuilder(Time.seconds(1))
                // 设置状态的更行类型1.OnCreateAndWrite（默认类型）：创建和写入的时候更新状态；OnReadAndWrite：除创建和写入，读的时候也可以更新状态
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                // 设置过期的状态是否还能读取，1.NeverReturnExpired（默认设置）：过期的状态不能再读取2.ReturnExpiredIfNotCleanedUp过期后，仍可以读取，在cleanup之前
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();

        ValueStateDescriptor<Tuple2<Long, Long>> descriptor = new ValueStateDescriptor<>(
                "average",
                TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
                }));

        descriptor.enableTimeToLive(stateTtlConfig);
        sum = getRuntimeContext().getState(descriptor);
    }
}

2.1 State Time-To-Live (TTL)

可以对键空状态设置TTL（time-to-live）,当状态生命周期结束会自动被清除。

所有集合类型状态都支持per-entry TTLs，意味着list和map类型的状态可以对单个元素或键值对设置TTL。
使用状态TTL必须首先构建StateTtlConfig配置对象，TTL功能可以通过状态描述符(state descriptor)传递。

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();
    
ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text state", String.class);
stateDescriptor.enableTimeToLive(ttlConfig);

TTL配置有以下几个选择性考虑：
首先newBuilder方法的参数是必须的，它设置TTL的值。
The update type configures when the state TTL is refreshed (by default OnCreateAndWrite):

StateTtlConfig.UpdateType.OnCreateAndWrite - only on creation and write access
StateTtlConfig.UpdateType.OnReadAndWrite - also on read access

The state visibility configures whether the expired value is returned on read access if it is not cleaned up yet (by default NeverReturnExpired):

StateTtlConfig.StateVisibility.NeverReturnExpired - expired value is never returned
StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp - returned if still available

NeverReturnExpired意味着过期的状态不会再被返回，即使其还没被清除。
ReturnExpiredIfNotCleanedUp 意味着即使过期了只要状态没有被清除，都会被返回。

注意：

1. 状态后端存储了最后一次修改的时间戳和用户值，意味着它增加了消耗状态的存储空间。堆状态后端存储一个Java对象以及其指向用户状态对象的索引和在内存中的原始长值。RocksDB状态后端每增加一个存储值、list 值、map键值对都会占用8字节的存储空间。
2. Only TTLs in reference to processing time are currently supported.
3. 当恢复状态，使用TTL设置窗台描述符，但没有体检配置TTL会褒奖用性错误和StateMigrationException异常。
4. TTL配置不是checkpoing或savepoint的一部分，而是操作运行job的一种方式。
5. map State目前支持null值但是该值的序列化必须能够处理null。如果序列化不支持null值，可能会用NullableSerializer包装，但是会消耗额外的内存。
   6.State TTL在python语言中还不支持。

2.2 Cleanup of Expired State

默认过期的值在读取的时候会被清除。但是可以设置StateTtlConfig使cleanup机制失效。

import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .disableCleanupInBackground()
    .build();

Flink提供了更细粒度状态清理机制

2.2.1 Cleanup in full snapshot

为了缩减状态所占空间，可以在获取到状态的快照后激活清理状态机制。在目前的实现中，除了可以清除本地的过期状态，其他的本地状态不会被清除。可以在StateTtlConfig中设置：

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .cleanupFullSnapshot()
    .build();

这种快照方式不能在RocksDB状态后端的增量checkpointing中使用。
对于已经存在的job，清除状态的策略可以在StateTtlConfig中任何时候被激活或暂停，例如从savepoint中恢复。

2.2.2 Incremental cleanup

另一个清除状态的策略是增量的清除状态的元素。激活该种策略可以在每次读取或写入数据时执行一次该清除策略。系统的state backend会持有所有状态的?个全局迭代器。每?次当??户访问状态，该迭代器就会增量迭代?个批次数据，检查是否存在过期的数据，如果存在就删除。

import org.apache.flink.api.common.state.StateTtlConfig;
 StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .cleanupIncrementally(10, true)
    .build();

该策略有两个参数，第一个是清除策略激活后每次检查几个条目，第二个参数定义是否每次处理记录时触发清除状态机制。默认每次检查5个条目不用每次处理记录时激活清除机制。
注意：

1. 如果没有读取状态或没有处理记录，过期的状态会被持久化。
2. 花费在增量清除状态上的时间将增加处理记录的延迟。
3. 目前增量清除机制只对堆状态后端有效。对于RocksDB是无效的。
4. 如果对状态后端和同步快照一起使用，当迭代的时候，全局迭代器会保存一个所有key的副本，因为其不支持同时修改。同时启动快照会增加内存消耗，如果时异步快照将不会有这种问题。
5. 对于已经存在的job，清除状态的策略可以在StateTtlConfig中任何时候被激活或暂停，例如从savepoint中恢复。

2.2.3 Cleanup during RocksDB compaction

如果?户使?的是RocksDB作为状态后端实现，?户可以在RocksDB在做Compation的时候加?Filter，对过期的数据进?检查。删除过期数据。

import org.apache.flink.api.common.state.StateTtlConfig;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(1))
    .cleanupInRocksdbCompactFilter(1000)
    .build();

每次处理一定数量的状态后，RocksDB compaction filter会查询目前的时间戳检查其是否过期。可以传递一个值改变这个值。增加更新时间戳的频率可以改善清除过期状态的速率但是将减少compaction的性能，因为它使?本地代码中的JNI调?，因此会降低压缩性能。默认该值时1000。
You can activate debug logs from the native code of RocksDB filter by activating debug level for FlinkCompactionFilter:

log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG
Notes:

1. 调用该TTL filter时，会降低compaction的速度，因为TTL filter 需要解析上次访问状态的时间戳和检查将要被compaction每个key的状态。对于集合状态的状态，检查时也会调用每个被存储的元素。
2. 如果这个特性和list state状态联合使用且其有不定长的元素，the native TTL filter has to call additionally a Flink java type serializer of the element over JNI per each state entry where at least the first element has expired to determine the offset of the next unexpired element。
3. 对于已经存在的job，清除状态的策略可以在StateTtlConfig中任何时候被激活或暂停，例如从savepoint中恢复。

3. State in the Scala DataStream API

除了上述描述的接口，Scala API 还有具有状态的map()或flatMap()方法，其具有ValueState状态，可以应用在keyedStream中。用户函数获取当前在Option中的ValueState的值然后返回一个更新的值区更新状态。

val stream: DataStream[(String, Int)] = ...

val counts: DataStream[(String, Int)] = stream
  .keyBy(_._1)
  .mapWithState((in: (String, Int), count: Option[Int]) =>
    count match {
      case Some(c) => ( (in._1, c), Some(c + in._2) )
      case None => ( (in._1, 0), Some(in._2) )
    })

mapWithState和flatMapWithState的区别：

1. map和flatMap的固有不同：前者是输入一个元素返回一个元素，后者则是输入一个元素可以返回0到多个元素

3. Operator State

算子状态（Operator State/ non-keyed state）是一种绑定在并行算子实例的状态。Kafka连接器是Flink算子状态的一个很好的应用示例。每一个并行的kafka消费者都会保存一个topic分区和offsets的映射作为算子状态。
算子状态支持分布式并行算子的并行度改变，有多种方案重新分配状态。
在Flink状态应用程序种一般不适用算子状态，它通常用在source/sink的应用场景种，因为没有key为状态进行分区。

4. Broadcast State

广播状态（Broadcast State）是一个特殊的算子状态。主要用来处理一条流被下游流的所有子任务所使用，它的状态可以被第二条流使用。有这样一个例子可以很好的解释广播流的使用场景：一个低吞吐量的流包含一系列评估来自另一个流的元素的规则。
广播流状态和算子状态区别在于算子流：

1.是一个map的格式
2.it is only available to specific operators that have as inputs a broadcasted stream and a non-broadcasted one, and
3.such an operator can have multiple broadcast states with different names.

5. Using Operator State

使用算子需要状态函数实现CheckpointFunction接口。

5.1 CheckpointedFunction

CheckpointedFunction 接口提供了多种分布式访问非键空状态方案，它需要实现以下两种方法：

void snapshotState(FunctionSnapshotContext context) throws Exception;

void initializeState(FunctionInitializationContext context) throws Exception;

当执行checkpoint就会调用snapshotState()方法。当状态函数第一次被初始化或状态函数从checkpoint恢复时相应的initializeState()方法。状态初始化或状态恢复都会调用initializeState()方法。

目前支持list类型的算子状态，算子状态应该是可以序列化对象的list，每个元素相互独立，因此可以重新分布于分布式计算中。基于获取状态的方法，有以下分布式方案可以选用：

1.Even-split redistribution(平均拆分):每个并行算子状态返回一个list类型状态，整个状态逻辑上由所有并行状态构成。在恢复或重新分配状态时，list被平均的分配到子任务中。每个算子维护一个sublist可以是空的list或者包含多个元素的list。例如，如果并行度是1，算子的检查点状态包含两个元素，当增加并行度为2时，这两个元素就会分别分配到子任务中。
2.Union redistribution：每个并行算子状态返回一个list类型状态，整个状态逻辑上由所有并行状态构成。在恢复或重新分配状态时，每个子任务获取一个完整的list，如果状态很大，最好不要这样使用。检查点元素据回为每个list entry保存一个offset，可能导致 RPC framesize or out-of-memory errors。

下面是一个 使用even-split redistribution 的示例，状态算子SinkFunction使用CheckpointedFunction先缓存一定的元素然后再将元素输出：

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import scala.Tuple2;
import java.util.ArrayList;
import java.util.List;

class BufferingSink implements SinkFunction<Tuple2<String, Integer>>, CheckpointedFunction {
    
    // 定义缓存多少条数据后开始sink，示例中用final但没有初始化值
    private int threshold;

    // 定义checkpoint状态
    private transient ListState<Tuple2<String, Integer>> checkpointState;

    // 用于缓存数据
    private List<Tuple2<String, Integer>> bufferedElements;

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

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

    @Override // checkpoint操作
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        checkpointState.clear();
        for (Tuple2<String, Integer> element : bufferedElements) {
            checkpointState.add(element);
        }
    }

    @Override // 初始化状态或恢复状态
    public void initializeState(FunctionInitializationContext context) throws Exception {
        
        ListStateDescriptor<Tuple2<String, Integer>> descriptor =
                new ListStateDescriptor<Tuple2<String, Integer>>(
                        "buffered-elements",
                        TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {
                        }));
        
        checkpointState = context.getOperatorStateStore().getListState(descriptor);

        if (context.isRestored()) {
            for (Tuple2<String, Integer> element : checkpointState.get()) {
                bufferedElements.add(element);
            }
        }
    }
}

initializeState方法的FunctionInitializationContext参数是初始化非键空状态的一个容器，当设置检查点时会将非键空状态对象存放于ListState类型的状态中。
PS：键空状态对应 KeyedStateStore getKeyedStateStore()方法。
算子状态的初始化和键空状态的初始化很相似，使用StateDescriptor对象包含状态名称和状态值类型的信息：

ListStateDescriptor<Tuple2<String, Integer>> descriptor =
    new ListStateDescriptor<>(
        "buffered-elements",
        TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));

checkpointedState = context.getOperatorStateStore().getListState(descriptor);

getListState(descriptor)，getUnionListState(descriptor)分别对应上面个介绍的两种重分配分区策略。
初始化状态容器后使用isRestored()方法检查是否需要错误恢复。

6. Stateful Source Functions

状态源算子相对其他算子状态要复杂。为了保证对状态和输出集合的原子性，必须从源上下文环境获取锁。

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

    /**  current offset for exactly once semantics */
    private Long offset = 0L;

    /** flag for job cancellation */
    private volatile boolean isRunning = true;
    
    /** Our state object. */
    private ListState<Long> state;

    @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 void initializeState(FunctionInitializationContext context) throws Exception {
        state = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
                "state",
                LongSerializer.INSTANCE));
                
        // restore any state that we might already have to our fields, initialize state
        // is also called in case of restore.
        for (Long l : state.get()) {
            offset = l;
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        state.clear();
        state.add(offset);
    }
}

详情参考官网