flinkflink

Working with State

2018-01-22  本文已影响77人  小C菜鸟

原文链接

Keyed State and Operator State

在Flink中有两种基本类型的状态：Keyed State and Operator State。

Keyed State

Keyed State总是和keys相关，并且只能用于KeyedStream上的函数和操作。
你可以将Keyed State认为是已经被分段或分区的Operator State，每个key都有且仅有一个state-partition。每个keyed-state逻辑上绑定到一个唯一的<parallel-operator-instance, key>组合上，并且由于每个key“属于”keyed operator的一个并行实例，所以我们可以简单的认为是<operator,key>。
Keyed State进一步被组织到所谓的Key Groups中。Key Groups是Flink能够重新分配keyed State的原子单元。Key Groups的数量等于定义的最大并行度。在一个keyed operator的并行实例执行期间，它与一个或多个Key Groups配合工作。

Operator State

对于Operator State(或者non-keyed state)，每个operator state绑定到一个并行operator实例上。在Flink中，Kafka Connector是一个使用Operator State的很好的例子。每个并行Kafka消费者实例维护一个主题分区和偏移的map作为它的Operator State。
当并行度被修改时，Operator State接口支持在并行operator实例上重新分配状态。进行这种重新分配可以有不同的方案。

Raw and Managed State

Keyed State 和 Operator State 有两种形式: managed和raw。
Managed State表示数据结构由Flink runtime控制，例如内部哈希表，或者RocksDB。例如，“ValueState”，“ListState”等等。Flink的runtime层编码State并将其写入checkpoint中。
Raw State是operator保存在它的数据结构中的state。当进行checkpoint时，它只写入字节序列到checkpoint中。Flink并不知道状态的数据结构，并且只能看到raw字节。
所有的数据流函数都可以使用managed state，但是raw state接口只可以在操作符的实现类中使用。推荐使用managed state(而不是raw state)，因为使用managed state，当并行度变化时，Flink可以自动的重新分布状态，并且可以做更好的内存管理。
注意 如果你的managed state需要自定义序列化逻辑，请参见managed state的自定义序列化以确保未来的兼容性。Flink默认的序列化不需要特殊处理。

使用Managed Keyed State

managed keyed state接口提供了对当前输入元素的key的不同类型的状态的访问。这意味着这种类型的状态只能在KeyedStream中使用，它可以通过stream.keyBy(...)创建。
现在，我们首先看下不同类型的状态，然后展示如何在程序中使用它们。可用的状态原语是：

这是一个显示了所有部分如何组合在一起的FlatMapFunction示例:

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();

// the printed output will be (1,4) and (1,5)

这个例子实现了一个计数窗口。我们以元组的第一个属性为key(在示例中都有相同的key 1)。该函数存储计数器和一个累加和到ValueState中。一旦计数器达到2，它会发出平均值并且清空状态以便重新从0开始。注意，如果我们在元组的第一个属性中有不同的值，那么将为每个不同的输入key保留不同的状态值。

State in the Scala DataStream API

除了上面描述的接口，Scala API在KeyedStream上为使用单个ValueState的有状态的map() 或 flatMap() 函数提供了快捷方式。用户函数在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) )
    })

Using Managed Operator State

为了使用managed operator state，有状态的函数可以实现更通用的CheckpointedFunction接口，或者ListCheckpointed<T extends Serializable>接口。

CheckpointedFunction

CheckpointedFunction接口提供了访问具备不同的重新分配策略的非keyed状态。它需要方式的实现:

void snapshotState(FunctionSnapshotContext context) throws Exception;

void initializeState(FunctionInitializationContext context) throws Exception;

每当要执行checkpoint时，都会调用snapshotState()方法。对应的 initializeState()在每次用户定义的函数初始化时调用，即函数第一次初始化或者函数从较早的checkpoint恢复时。因此initializeState()不仅是不同类型的状态初始化的地方，也是包含恢复逻辑的地方。
目前，支持列表风格的managed操作符状态。状态期望是一个可序列化对象的列表，每个元素都是独立的，因此可以在弹性扩容时重新分配。换句话说，这些对象是非keyed状态可重新分配的最佳粒度。根据状态访问方法，定义了下属重新分配方案：

initializeState方法接受FunctionInitializationContext作为参数。它用来初始化非keyed状态“容器”。上面是ListState类型的容器，当进行checkpoint时非keyed状态的对象存储在ListState中。
注意状态是如何初始化的，类似于keyed状态，有一个包含状态的名称和状态所持有的状态的信息的StateDescriptor：

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

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

状态访问方法的命名约定包含它的状态结构的重新分配的模式。例如，在恢复时使用Union redistribution方案的list state，通过使用getUnionListState(descriptor)方法访问状态。如果方法名不包含重新分配模式，例如getListState(descriptor)，它意味着重新分配方案使用基本的even-split redistribution。
初始化容器后，我们使用context的isRestored()方法来检查我们是否正在从故障中恢复。如果是true，也就是正在恢复中，则应用恢复逻辑。
就像BufferingSink代码中所示，在状态初始化时恢复的ListState保存在一个类变量中，以便snapshotState()中使用。ListState清除所有前一个checkpoint包含的所有对象，然后填充我们想要checkpoint的新对象。
另外，keyed状态也能在 initializeState() 方法中初始化。这通过使用提供的FunctionInitializationContext实现。

ListCheckpointed

ListCheckpointed接口是CheckpointedFunction的限制更严的变体，它只支持恢复时使用even-split redistribution方案的列表风格的状态。它也要求实现两个方法：

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

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

在snapshotState()上操作符应该返回一个checkpoint的对象列表，并且恢复时restoreState必须处理这样一个列表。如果状态是不可分割的，你可以在snapshotState()上总是返回Collections.singletonList(MY_STATE)。

Stateful Source Functions

有状态的Source相比其它操作符需要关注多一点。为了保证状态和输出集合的更新是原子的(精确一次语义在故障/恢复时要求)，用户要求从Source的context中获取锁。

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);
    }

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

一些操作符当一个checkpoint被Flink完全确认时可能需要与外部世界通信。在这种情况下见org.apache.flink.runtime.state.CheckpointListener接口。

上一篇 下一篇

猜你喜欢

热点阅读