devbox
煮酒与君饮
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink:状态与容错_reducingstatedescriptor

作者：煮酒与君饮 | 2024-08-23 03:00:03

赞

踩

reducingstatedescriptor

一、Keyed State

1、用途

Keyed State接口目前有5种不同类型状态的访问接口，这些状态都作用于当前输入数据的key下。换句话说，这些状态只能用在keyedStream上，可以通过stream.keyBy(…)得到KeyedStream.

这5种接口分别如下：
1、ValueState：保存一个可以更新和检索的值
2、ListState: 保存一个元素的列表
3、ReducingState: 保存一个单值，表示添加到状态的所有值的聚合
4、AggregatingState<IN, OUT>: 保留一个单值，表示添加到状态的所有值的聚合
5、MapState<UK, UV>: 维护了一个映射列表
接口继承关系如图:（其中AppendingState源码在ReducingState源码部分展示）
在这里插入图片描述

这些状态对象仅用于与状态交互。状态本身不一定存储在内存中，还可能在磁盘或其他位置。 另外需要牢记的是从状态中获取的值取决于输入元素所代表的 key。 因此，在不同 key 上调用同一个接口，可能得到不同的值。
你必须创建一个 StateDescriptor，才能得到对应的状态句柄。 这保存了状态名称（正如我们稍后将看到的，你可以创建多个状态，并且它们必须具有唯一的名称以便可以引用它们）， 状态所持有值的类型，并且可能包含用户指定的函数，例如ReduceFunction。 根据不同的状态类型，可以创建ValueStateDescriptor，ListStateDescriptor， ReducingStateDescriptor 或 MapStateDescriptor。
状态通过 RuntimeContext 进行访问，因此只能在 rich functions 中使用。

RichFunction 中 RuntimeContext 提供如下方法：
1、ValueState getState(ValueStateDescriptor)
2、ReducingState getReducingState(ReducingStateDescriptor)
3、ListState getListState(ListStateDescriptor)
4、AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)
5、MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)

2、源码
1、ValueState和ValueStateDescriptor

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;

import java.io.IOException;

/**
 * {@link State} interface for partitioned single-value state. The value can be retrieved or
 * updated.
 * {@link State} 用于分区单值状态的接口。 可以检索或更新该值。
 *
 *
 * <p>The state is accessed and modified by user functions, and checkpointed consistently
 * by the system as part of the distributed snapshots.
 * 状态由用户函数访问和修改，并由系统作为分布式快照的一部分一致地设置检查点。
 *
 * <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.
 * 该状态只能由应用于 {@code KeyedStream} 的函数访问。 键是系统自动提供的，所以函数总是看到映射到当前元素键的值。
 * 这样，系统就可以一致地处理流和状态分区。
 *
 * @param <T> Type of the value in the state. 状态中值的类型。
 */
@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)}.
	 * 如果您在创建 {@link ValueStateDescriptor} 时未指定默认值，则在之前使用 {@link #update(Object)} 设置的值时，将返回 {@code null}。
	 *
	 * @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.
	 * 将 {@link #value()} 可访问的运算符状态更新为给定的
	 价值。 下次调用 {@link #value()} 时（对于相同的状态
	 分区）返回的状态将代表更新后的值。 当一个
	 分区状态更新为 null，当前键的状态将被删除，并在下次访问时返回默认值。
	 *
	 * @param value The new value for the state.
	 *              状态的新值。
	 *
	 * @throws IOException Thrown if the system cannot access the state.
	 */
	void update(T value) throws IOException;

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;

/**
 * {@link StateDescriptor} for {@link ValueState}. This can be used to create partitioned
 * value state using
 * {@link StateDescriptor} 用于 {@link ValueState}。 这可用于创建分区值状态使用
 * {@link org.apache.flink.api.common.functions.RuntimeContext#getState(ValueStateDescriptor)}.
 *
 * <p>If you don't use one of the constructors that set a default value the value that you
 * get when reading a {@link ValueState} using {@link ValueState#value()} will be {@code null}.
 * 如果您不使用设置默认值的构造函数之一，则使用 {@link ValueState#value()} 读取 {@link ValueState} 时获得的值将是 {@code null}。
 *
 * @param <T> The type of the values that the value state can hold.
 *           值状态可以保存的值的类型。
 */
@PublicEvolving
public class ValueStateDescriptor<T> extends StateDescriptor<ValueState<T>, T> {

	private static final long serialVersionUID = 1L;

	/**
	 * Creates a new {@code ValueStateDescriptor} with the given name and type
	 * 使用给定的名称和类型创建一个新的 {@code ValueStateDescriptor}
	 *
	 * <p>If this constructor fails (because it is not possible to describe the type via a class),
	 * consider using the {@link #ValueStateDescriptor(String, TypeInformation)} constructor.
	 * 如果此构造函数失败（因为无法通过类来描述类型），请考虑使用 {@link #ValueStateDescriptor(String, TypeInformation)} 构造函数。
	 *
	 * @param name The (unique) name for the state.状态的（唯一）名称。
	 * @param typeClass The type of the values in the state.状态中值的类型。
	 */
	public ValueStateDescriptor(String name, Class<T> typeClass) {
		super(name, typeClass, null);
	}

	/**
	 * Creates a new {@code ValueStateDescriptor} with the given name and type.
	 * 使用给定的名称和类型创建一个新的 {@code ValueStateDescriptor}。
	 *
	 * @param name The (unique) name for the state.
	 * @param typeInfo The type of the values in the state.
	 */
	public ValueStateDescriptor(String name, TypeInformation<T> typeInfo) {
		super(name, typeInfo, null);
	}

	/**
	 * Creates a new {@code ValueStateDescriptor} with the given name and the specific serializer.
	 * 使用给定的名称和特定的序列化程序创建一个新的 {@code ValueStateDescriptor}。
	 *
	 * @param name The (unique) name for the state.
	 * @param typeSerializer The type serializer of the values in the state.
	 *                       状态中值的类型序列化器。
	 */
	public ValueStateDescriptor(String name, TypeSerializer<T> typeSerializer) {
		super(name, typeSerializer, null);
	}

	@Override
	public Type getType() {
		return Type.VALUE;
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67

测试案例如下：

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(value -> value.f0)
        .flatMap(new CountWindowAverage())
        .print();

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
2、ListState和ListStateDescriptor

接口ListState继承自接口MergingState和AppendingState
在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;

import java.util.List;

/**
 * {@link State} interface for partitioned list state in Operations.
 * The state is accessed and modified by user functions, and checkpointed consistently
 * by the system as part of the distributed snapshots.
 * {@link State} 操作中分区列表状态的接口。 状态由用户函数访问和修改，并由系统作为分布式快照的一部分一致地设置检查点。
 *
 * <p>The state can be a keyed list state or an operator list state.
 * 该状态可以是键控列表状态或操作员列表状态。
 *
 * <p>When it is a keyed list state, it is accessed 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.
 * 当它是键控列表状态时，它由应用于 {@code KeyedStream} 的函数访问。 键是系统自动提供的，所以函数总是看到映射到当前元素键的值。
 * 这样，系统就可以一致地处理流和状态分区。
 *
 * <p>When it is an operator list state, the list is a collection of state items that are
 * independent from each other and eligible for redistribution across operator instances in case
 * of changed operator parallelism.
 * 当它是一个算子列表状态时，该列表是一个状态项的集合，这些状态项彼此独立，并且在算子并行度发生变化时可以在算子实例之间重新分配。
 *
 * @param <T> Type of values that this list state keeps. 此列表状态保留的值的类型。
 */
@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.
	 * 通过将现有值更新为给定的值列表来更新 {@link #get()} 可访问的运算符状态。
	 * 下次调用 {@link #get()} 时（对于相同的状态分区），返回的状态将代表更新后的列表。
	 *
	 * <p>If null or an empty list is passed in, the state value will be null.
	 * 如果传入 null 或空列表，则状态值为 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.
	 * 通过将给定值添加到现有值列表来更新 {@link #get()} 可访问的运算符状态。
	 * 下次调用 {@link #get()} 时（对于相同的状态分区），返回的状态将代表更新后的列表。
	 *
	 * <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;
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.ListSerializer;
import org.apache.flink.api.java.typeutils.ListTypeInfo;

import java.util.List;

/**
 * A {@link StateDescriptor} for {@link ListState}. This can be used to create state where the type
 * is a list that can be appended and iterated over.
 * {@link ListState} 的 {@link StateDescriptor}。 这可用于创建类型是可以附加和迭代的列表的状态。
 *
 * <p>Using {@code ListState} is typically more efficient than manually maintaining a list in a
 * {@link ValueState}, because the backing implementation can support efficient appends, rather than
 * replacing the full list on write.
 * 使用 {@code ListState} 通常比在 {@link ValueState} 中手动维护列表更有效，
 * 因为支持实现可以支持高效的追加，而不是在写入时替换完整列表。
 *
 * <p>To create keyed list state (on a KeyedStream), use
 * {@link org.apache.flink.api.common.functions.RuntimeContext#getListState(ListStateDescriptor)}.
 *
 * @param <T> The type of the values that can be added to the list state.
 */
@PublicEvolving
public class ListStateDescriptor<T> extends StateDescriptor<ListState<T>, List<T>> {
	private static final long serialVersionUID = 2L;

	/**
	 * Creates a new {@code ListStateDescriptor} with the given name and list element type.
	 *
	 * <p>If this constructor fails (because it is not possible to describe the type via a class),
	 * consider using the {@link #ListStateDescriptor(String, TypeInformation)} constructor.
	 * 如果此构造函数失败（因为无法通过类来描述类型），请考虑使用 {@link #ListStateDescriptor(String, TypeInformation)} 构造函数。
	 *
	 * @param name The (unique) name for the state.
	 * @param elementTypeClass The type of the elements in the state.
	 */
	public ListStateDescriptor(String name, Class<T> elementTypeClass) {
		super(name, new ListTypeInfo<>(elementTypeClass), null);
	}

	/**
	 * Creates a new {@code ListStateDescriptor} with the given name and list element type.
	 *
	 * @param name The (unique) name for the state.
	 * @param elementTypeInfo The type of the elements in the state.
	 */
	public ListStateDescriptor(String name, TypeInformation<T> elementTypeInfo) {
		super(name, new ListTypeInfo<>(elementTypeInfo), null);
	}

	/**
	 * Creates a new {@code ListStateDescriptor} with the given name and list element type.
	 *
	 * @param name The (unique) name for the state.
	 * @param typeSerializer The type serializer for the list values.列表值的类型序列化程序。
	 */
	public ListStateDescriptor(String name, TypeSerializer<T> typeSerializer) {
		super(name, new ListSerializer<>(typeSerializer), null);
	}

	/**
	 * Gets the serializer for the elements contained in the list.
	 * 获取列表中包含的元素的序列化程序。
	 *
	 * @return The serializer for the elements in the list.
	 */
	public TypeSerializer<T> getElementSerializer() {
		// call getSerializer() here to get the initialization check and proper error message
		final TypeSerializer<List<T>> rawSerializer = getSerializer();
		if (!(rawSerializer instanceof ListSerializer)) {
			throw new IllegalStateException();
		}

		return ((ListSerializer<T>) rawSerializer).getElementSerializer();
	}

	@Override
	public Type getType() {
		return Type.LIST;
	}
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
3、ReducingState和ReducingStateDescriptor

接口ReducingState继承自接口MergingState和AppendingState

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;

/**
 * {@link State} interface for reducing state. Elements can be added to the state, they will
 * be combined using a reduce function. The current state can be inspected.
 * {@link State} 接口用于减少状态。 元素可以添加到状态中，它们将使用 reduce 函数进行组合。 可以检查当前状态。
 *
 * <p>The state is accessed and modified by user functions, and checkpointed consistently
 * by the system as part of the distributed snapshots.
 *
 * <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.
 *
 * @param <T> Type of the value in the operator state
 */
@PublicEvolving
public interface ReducingState<T> extends MergingState<T, T> {}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;

/**
 * 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.
 * 允许合并状态的 {@link AppendingState} 扩展。
 * 也就是说，可以将 {@link MergingState} 的两个实例合并为一个实例，该实例包含两个合并状态的所有信息。
 *
 * @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> { }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;

/**
 * Base interface for partitioned state that supports adding elements and inspecting the current
 * state. Elements can either be kept in a buffer (list-like) or aggregated into one value.
 * 支持添加元素和检查当前状态的分区状态的基本接口。 元素既可以保存在缓冲区中（类似于列表），也可以聚合为一个值。
 *
 * <p>The state is accessed and modified by user functions, and checkpointed consistently
 * by the system as part of the distributed snapshots.
 *
 * <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.
 *
 * @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 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.
	 * 通过将给定值添加到值列表来更新 {@link #get()} 可访问的运算符状态。
	 * 下次调用 {@link #get()} 时（对于相同的状态分区），返回的状态将代表更新后的列表。
	 *
	 * <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;

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * {@link StateDescriptor} for {@link ReducingState}. This can be used to create partitioned
 * reducing state using
 * {@link org.apache.flink.api.common.functions.RuntimeContext#getReducingState(ReducingStateDescriptor)}.
 *
 * @param <T> The type of the values that can be added to the list state.
 */
@PublicEvolving
public class ReducingStateDescriptor<T> extends StateDescriptor<ReducingState<T>, T> {

	private static final long serialVersionUID = 1L;

	private final ReduceFunction<T> reduceFunction;

	/**
	 * Creates a new {@code ReducingStateDescriptor} with the given name, type, and default value.
	 *
	 * <p>If this constructor fails (because it is not possible to describe the type via a class),
	 * consider using the {@link #ReducingStateDescriptor(String, ReduceFunction, TypeInformation)} constructor.
	 *
	 * @param name The (unique) name for the state.
	 * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.
	 * @param typeClass The type of the values in the state.
	 */
	public ReducingStateDescriptor(String name, ReduceFunction<T> reduceFunction, Class<T> typeClass) {
		super(name, typeClass, null);
		this.reduceFunction = checkNotNull(reduceFunction);

		if (reduceFunction instanceof RichFunction) {
			throw new UnsupportedOperationException("ReduceFunction of ReducingState can not be a RichFunction.");
		}
	}

	/**
	 * Creates a new {@code ReducingStateDescriptor} with the given name and default value.
	 *
	 * @param name The (unique) name for the state.
	 * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.
	 * @param typeInfo The type of the values in the state.
	 */
	public ReducingStateDescriptor(String name, ReduceFunction<T> reduceFunction, TypeInformation<T> typeInfo) {
		super(name, typeInfo, null);
		this.reduceFunction = checkNotNull(reduceFunction);
	}

	/**
	 * Creates a new {@code ValueStateDescriptor} with the given name and default value.
	 *
	 * @param name The (unique) name for the state.
	 * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.
	 * @param typeSerializer The type serializer of the values in the state.
	 */
	public ReducingStateDescriptor(String name, ReduceFunction<T> reduceFunction, TypeSerializer<T> typeSerializer) {
		super(name, typeSerializer, null);
		this.reduceFunction = checkNotNull(reduceFunction);
	}

	/**
	 * Returns the reduce function to be used for the reducing state.
	 */
	public ReduceFunction<T> getReduceFunction() {
		return reduceFunction;
	}

	@Override
	public Type getType() {
		return Type.REDUCING;
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
4、AggregatingState和AggregatingStateDescriptor

接口AggregatingState也是继承自接口MergingState和AppendingState，和 ReducingState 相反的是, 聚合类型可能与 添加到状态的元素的类型不同。

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * {@link State} interface for aggregating state, based on an
 * {@link AggregateFunction}. Elements that are added to this type of state will
 * be eagerly pre-aggregated using a given {@code AggregateFunction}.
 * {@link State} 用于聚合状态的接口，基于 {@link AggregateFunction}。
 * 添加到此类状态的元素将使用给定的 {@code AggregateFunction} 进行预聚合。
 *
 * <p>The state holds internally always the accumulator type of the {@code AggregateFunction}.
 * When accessing the result of the state, the function's
 * {@link AggregateFunction#getResult(Object)} method.
 * 状态在内部始终保持 {@code AggregateFunction} 的累加器类型。
 * 访问状态结果时，函数的 {@link AggregateFunction#getResult(Object)} 方法。
 *
 * <p>The state is accessed and modified by user functions, and checkpointed consistently
 * by the system as part of the distributed snapshots.
 *
 * <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.
 *
 * @param <IN> Type of the value added to the state.
 * @param <OUT> Type of the value extracted from the state.
 */
@PublicEvolving
public interface AggregatingState<IN, OUT> extends MergingState<IN, OUT> {}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * A StateDescriptor for {@link AggregatingState}.
 *
 * <p>The type internally stored in the state is the type of the {@code Accumulator} of the
 * {@code AggregateFunction}.
 *
 * @param <IN> The type of the values that are added to the state.
 * @param <ACC> The type of the accumulator (intermediate aggregation state).
 * @param <OUT> The type of the values that are returned from the state.
 */
@PublicEvolving
public class AggregatingStateDescriptor<IN, ACC, OUT> extends StateDescriptor<AggregatingState<IN, OUT>, ACC> {
	private static final long serialVersionUID = 1L;

	/** The aggregation function for the state. */
	private final AggregateFunction<IN, ACC, OUT> aggFunction;

	/**
	 * Creates a new state descriptor with the given name, function, and type.
	 *
	 * <p>If this constructor fails (because it is not possible to describe the type via a class),
	 * consider using the {@link #AggregatingStateDescriptor(String, AggregateFunction, TypeInformation)} constructor.
	 *
	 * @param name The (unique) name for the state.
	 * @param aggFunction The {@code AggregateFunction} used to aggregate the state.
	 * @param stateType The type of the accumulator. The accumulator is stored in the state.
	 */
	public AggregatingStateDescriptor(
			String name,
			AggregateFunction<IN, ACC, OUT> aggFunction,
			Class<ACC> stateType) {

		super(name, stateType, null);
		this.aggFunction = checkNotNull(aggFunction);
	}

	/**
	 * Creates a new {@code ReducingStateDescriptor} with the given name and default value.
	 *
	 * @param name The (unique) name for the state.
	 * @param aggFunction The {@code AggregateFunction} used to aggregate the state.
	 * @param stateType The type of the accumulator. The accumulator is stored in the state.
	 */
	public AggregatingStateDescriptor(
			String name,
			AggregateFunction<IN, ACC, OUT> aggFunction,
			TypeInformation<ACC> stateType) {

		super(name, stateType, null);
		this.aggFunction = checkNotNull(aggFunction);
	}

	/**
	 * Creates a new {@code ValueStateDescriptor} with the given name and default value.
	 *
	 * @param name The (unique) name for the state.
	 * @param aggFunction The {@code AggregateFunction} used to aggregate the state.
	 * @param typeSerializer The serializer for the accumulator. The accumulator is stored in the state.
	 */
	public AggregatingStateDescriptor(
			String name,
			AggregateFunction<IN, ACC, OUT> aggFunction,
			TypeSerializer<ACC> typeSerializer) {

		super(name, typeSerializer, null);
		this.aggFunction = checkNotNull(aggFunction);
	}

	/**
	 * Returns the aggregate function to be used for the state.
	 */
	public AggregateFunction<IN, ACC, OUT> getAggregateFunction() {
		return aggFunction;
	}

	@Override
	public Type getType() {
		return Type.AGGREGATING;
	}
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
5、MapState和MapStateDescriptor

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;

import java.util.Iterator;
import java.util.Map;

/**
 * {@link State} interface for partitioned key-value state. The key-value pair can be
 * added, updated and retrieved.
 * {@link State} 用于分区键值状态的接口。 可以添加、更新和检索键值对。
 *
 * <p>The state is accessed and modified by user functions, and checkpointed consistently
 * by the system as part of the distributed snapshots.
 *
 * <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.
 *
 * @param <UK> Type of the keys in the state.
 * @param <UV> Type of the values in the state.
 */
@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;

	/**
	 * Returns true if this state contains no key-value mappings, otherwise false.
	 *
	 * @return True if this state contains no key-value mappings, otherwise false.
	 * 如果此状态不包含键值映射，则为真，否则为假。
	 *
	 * @throws Exception Thrown if the system cannot access the state.
	 */
	boolean isEmpty() throws Exception;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127

在这里插入图片描述

package org.apache.flink.api.common.state;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.MapSerializer;
import org.apache.flink.api.java.typeutils.MapTypeInfo;

import java.util.Map;

/**
 * A {@link StateDescriptor} for {@link MapState}. This can be used to create state where the type
 * is a map that can be updated and iterated over.
 *
 * <p>Using {@code MapState} is typically more efficient than manually maintaining a map in a
 * {@link ValueState}, because the backing implementation can support efficient updates, rather then
 * replacing the full map on write.
 *
 * <p>To create keyed map state (on a KeyedStream), use
 * {@link org.apache.flink.api.common.functions.RuntimeContext#getMapState(MapStateDescriptor)}.
 *
 * <p>Note: The map state with TTL currently supports {@code null} user values
 * only if the user value serializer can handle {@code null} values.
 * If the serializer does not support {@code null} values,
 * it can be wrapped with {@link org.apache.flink.api.java.typeutils.runtime.NullableSerializer}
 * at the cost of an extra byte in the serialized form.
 *
 * @param <UK> The type of the keys that can be added to the map state.
 */
@PublicEvolving
public class MapStateDescriptor<UK, UV> extends StateDescriptor<MapState<UK, UV>, Map<UK, UV>> {

	private static final long serialVersionUID = 1L;

	/**
	 * Create a new {@code MapStateDescriptor} with the given name and the given type serializers.
	 *
	 * @param name The name of the {@code MapStateDescriptor}.
	 * @param keySerializer The type serializer for the keys in the state.
	 * @param valueSerializer The type serializer for the values in the state.
	 */
	public MapStateDescriptor(String name, TypeSerializer<UK> keySerializer, TypeSerializer<UV> valueSerializer) {
		super(name, new MapSerializer<>(keySerializer, valueSerializer), null);
	}

	/**
	 * Create a new {@code MapStateDescriptor} with the given name and the given type information.
	 *
	 * @param name The name of the {@code MapStateDescriptor}.
	 * @param keyTypeInfo The type information for the keys in the state.
	 * @param valueTypeInfo The type information for the values in the state.
	 */
	public MapStateDescriptor(String name, TypeInformation<UK> keyTypeInfo, TypeInformation<UV> valueTypeInfo) {
		super(name, new MapTypeInfo<>(keyTypeInfo, valueTypeInfo), null);
	}

	/**
	 * Create a new {@code MapStateDescriptor} with the given name and the given type information.
	 *
	 * <p>If this constructor fails (because it is not possible to describe the type via a class),
	 * consider using the {@link #MapStateDescriptor(String, TypeInformation, TypeInformation)} constructor.
	 *
	 * @param name The name of the {@code MapStateDescriptor}.
	 * @param keyClass The class of the type of keys in the state.
	 * @param valueClass The class of the type of values in the state.
	 */
	public MapStateDescriptor(String name, Class<UK> keyClass, Class<UV> valueClass) {
		super(name, new MapTypeInfo<>(keyClass, valueClass), null);
	}

	@Override
	public Type getType() {
		return Type.MAP;
	}

	/**
	 * Gets the serializer for the keys in the state.
	 *
	 * @return The serializer for the keys in the state.
	 */
	public TypeSerializer<UK> getKeySerializer() {
		final TypeSerializer<Map<UK, UV>> rawSerializer = getSerializer();
		if (!(rawSerializer instanceof MapSerializer)) {
			throw new IllegalStateException("Unexpected serializer type.");
		}

		return ((MapSerializer<UK, UV>) rawSerializer).getKeySerializer();
	}

	/**
	 * Gets the serializer for the values in the state.
	 *
	 * @return The serializer for the values in the state.
	 */
	public TypeSerializer<UV> getValueSerializer() {
		final TypeSerializer<Map<UK, UV>> rawSerializer = getSerializer();
		if (!(rawSerializer instanceof MapSerializer)) {
			throw new IllegalStateException("Unexpected serializer type.");
		}

		return ((MapSerializer<UK, UV>) rawSerializer).getValueSerializer();
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103

二、Operator State

1、用途

Operator State与Keyed State的区别在于，Operator State与并行运算符实例有关系，每个并行运算符都对应一个独立的状态管理器；而Keyed State与流的key有关系，无论有多少个并行运算符，Keyed State的状态管理器的数量只与key的数量相同。

Operator State有两种类型的状态访问接口：
1、ListState
2、BroadcastState

当然，ListState和BroadcastState接口只是用于存储状态数据的数据结构，更好的使用operator state还得实现CheckpointedFunction接口：
CheckpointedFunction 接口提供了访问 non-keyed state 的方法，需要实现如下两个方法：

void snapshotState(FunctionSnapshotContext context) throws Exception;

void initializeState(FunctionInitializationContext context) throws Exception;

  • 1
  • 2
  • 3

进行 checkpoint 时会调用 snapshotState()。 用户自定义函数初始化时会调用 initializeState()，初始化包括第一次自定义函数初始化和从之前的 checkpoint 恢复。 因此 initializeState() 不仅是定义不同状态类型初始化的地方，也需要包括状态恢复的逻辑。

2、源码
1、CheckpointedFunction

在这里插入图片描述

package org.apache.flink.streaming.api.checkpoint;

import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.KeyedStateStore;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;

/**
 * This is the core interface for <i>stateful transformation functions</i>, meaning functions
 * that maintain state across individual stream records.
 * While more lightweight interfaces exist as shortcuts for various types of state, this interface offer the
 * greatest flexibility in managing both <i>keyed state</i> and <i>operator state</i>.
 * 这是<i>有状态转换函数</i>的核心接口，这意味着在各个流记录中维护状态的函数。
 * 虽然存在更多轻量级接口作为各种类型状态的快捷方式，但该接口在管理<i>键控状态</i>和<i>操作员状态</i>方面提供了最大的灵活性。
 *
 * <p>The section <a href="#shortcuts">Shortcuts</a> illustrates the common lightweight
 * ways to setup stateful functions typically used instead of the full fledged
 * abstraction represented by this interface.
 * <a href="#shortcuts">Shortcuts</a> 部分说明了设置有状态函数的常用轻量级方法，这些方法通常用于代替此接口表示的完整抽象。
 *
 * <h1>Initialization</h1>
 * The {@link CheckpointedFunction#initializeState(FunctionInitializationContext)} is called when
 * the parallel instance of the transformation function is created during distributed execution.
 * The method gives access to the {@link FunctionInitializationContext} which in turn gives access
 * to the to the {@link OperatorStateStore} and {@link KeyedStateStore}.
 * {@link CheckpointedFunction#initializeState(FunctionInitializationContext)} 在分布式执行期间创建转换函数的并行实例时被调用。
 * 该方法可以访问 {@link FunctionInitializationContext}，而后者又可以访问 {@link OperatorStateStore} 和 {@link KeyedStateStore}。
 *
 * <p>The {@code OperatorStateStore} and {@code KeyedStateStore} give access to the data structures
 * in which state should be stored for Flink to transparently manage and checkpoint it, such as
 * {@link org.apache.flink.api.common.state.ValueState} or
 * {@link org.apache.flink.api.common.state.ListState}.
 * {@code OperatorStateStore} 和 {@code KeyedStateStore} 允许访问应存储状态的数据结构，以便 Flink 透明地管理和检查它，
 * 例如 {@link org.apache.flink.api.common.state.ValueState} 或 {@link org.apache.flink.api.common.state.ListState}。
 *
 * <p><b>Note:</b> The {@code KeyedStateStore} can only be used when the transformation supports
 * <i>keyed state</i>, i.e., when it is applied on a keyed stream (after a {@code keyBy(...)}).
 * {@code KeyedStateStore} 只能在转换支持 <i>keyed state</i> 时使用，即当它应用于键控流时（在 {@code keyBy(...)} 之后）。
 *
 * <h1>Snapshot</h1>
 * The {@link CheckpointedFunction#snapshotState(FunctionSnapshotContext)} is called whenever a
 * checkpoint takes a state snapshot of the transformation function. Inside this method, functions typically
 * make sure that the checkpointed data structures (obtained in the initialization phase) are up
 * to date for a snapshot to be taken. The given snapshot context gives access to the metadata
 * of the checkpoint.
 * 每当检查点获取转换函数的状态快照时，就会调用 {@link CheckpointedFunction#snapshotState(FunctionSnapshotContext)}。
 * 在此方法中，函数通常确保检查点数据结构（在初始化阶段获得）是最新的，以便拍摄快照。 给定的快照上下文允许访问检查点的元数据。
 *
 * <p>In addition, functions can use this method as a hook to flush/commit/synchronize with
 * external systems.
 * 此外，函数可以使用此方法作为挂钩来刷新/提交/与外部系统同步。
 *
 * <h1>Example</h1>
 * The code example below illustrates how to use this interface for a function that keeps counts
 * of events per key and per parallel partition (parallel instance of the transformation function
 * during distributed execution).
 * 下面的代码示例说明了如何将此接口用于保持每个键和每个并行分区（分布式执行期间转换函数的并行实例）的事件计数的函数。
 * The example also changes of parallelism, which affect the count-per-parallel-partition by
 * adding up the counters of partitions that get merged on scale-down. Note that this is a
 * toy example, but should illustrate the basic skeleton for a stateful function.
 * 该示例还更改了并行性，它通过将按比例缩小合并的分区的计数器相加来影响每个并行分区的计数。 请注意，这是一个玩具示例，但应该说明有状态函数的基本框架。
 *
 * <p><pre>{@code
 * public class MyFunction<T> implements MapFunction<T, T>, CheckpointedFunction {
 *
 *     private ReducingState<Long> countPerKey;
 *     private ListState<Long> countPerPartition;
 *
 *     private long localCount;
 *
 *     public void initializeState(FunctionInitializationContext context) throws Exception {
 *         // get the state data structure for the per-key state
 *         countPerKey = context.getKeyedStateStore().getReducingState(
 *                 new ReducingStateDescriptor<>("perKeyCount", new AddFunction<>(), Long.class));
 *
 *         // get the state data structure for the per-partition state
 *         countPerPartition = context.getOperatorStateStore().getOperatorState(
 *                 new ListStateDescriptor<>("perPartitionCount", Long.class));
 *
 *         // initialize the "local count variable" based on the operator state
 *         for (Long l : countPerPartition.get()) {
 *             localCount += l;
 *         }
 *     }
 *
 *     public void snapshotState(FunctionSnapshotContext context) throws Exception {
 *         // the keyed state is always up to date anyways
 *         // just bring the per-partition state in shape
 *         countPerPartition.clear();
 *         countPerPartition.add(localCount);
 *     }
 *
 *     public T map(T value) throws Exception {
 *         // update the states
 *         countPerKey.add(1L);
 *         localCount++;
 *
 *         return value;
 *     }
 * }
 * }</pre>
 *
 * <hr>
 *
 * <h1><a name="shortcuts">Shortcuts</a></h1>
 * There are various ways that transformation functions can use state without implementing the
 * full-fledged {@code CheckpointedFunction} interface:
 * 转换函数可以通过多种方式使用状态，而无需实现成熟的 {@code CheckpointedFunction} 接口：
 *
 * <h4>Operator State</h4>
 * Checkpointing some state that is part of the function object itself is possible in a simpler way
 * by directly implementing the {@link ListCheckpointed} interface.
 * 通过直接实现 {@link ListCheckpointed} 接口，可以以更简单的方式检查属于函数对象本身的某些状态。
 *
 * <h4>Keyed State</h4>
 * Access to keyed state is possible via the {@link RuntimeContext}'s methods:
 * 可以通过 {@link RuntimeContext} 的方法访问键控状态
 * <pre>{@code
 * public class CountPerKeyFunction<T> extends RichMapFunction<T, T> {
 *
 *     private ValueState<Long> count;
 *
 *     public void open(Configuration cfg) throws Exception {
 *         count = getRuntimeContext().getState(new ValueStateDescriptor<>("myCount", Long.class));
 *     }
 *
 *     public T map(T value) throws Exception {
 *         Long current = count.get();
 *         count.update(current == null ? 1L : current + 1);
 *
 *         return value;
 *     }
 * }
 * }</pre>
 *
 * @see ListCheckpointed
 * @see RuntimeContext
 */
@Public
public interface CheckpointedFunction {

	/**
	 * This method is called when a snapshot for a checkpoint is requested. This acts as a hook to the function to
	 * ensure that all state is exposed by means previously offered through {@link FunctionInitializationContext} when
	 * the Function was initialized, or offered now by {@link FunctionSnapshotContext} itself.
	 * 当请求检查点的快照时调用此方法。 这充当函数的钩子，以确保所有状态都通过先前在函数初始化时通过 {@link FunctionInitializationContext} 提供的方式公开，
	 * 或者现在由 {@link FunctionSnapshotContext} 本身提供。
	 *
	 * @param context the context for drawing a snapshot of the operator
	 *                绘制操作员快照的上下文
	 * @throws Exception Thrown, if state could not be created ot restored.
	 */
	void snapshotState(FunctionSnapshotContext context) throws Exception;

	/**
	 * This method is called when the parallel function instance is created during distributed
	 * execution. Functions typically set up their state storing data structures in this method.
	 * 在分布式执行期间创建并行函数实例时调用此方法。 函数通常在此方法中设置其状态存储数据结构。
	 *
	 * @param context the context for initializing the operator
	 *                初始化操作符的上下文
	 * @throws Exception Thrown, if state could not be created ot restored.
	 */
	void initializeState(FunctionInitializationContext context) throws Exception;
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151
  • 152
  • 153
  • 154
  • 155
  • 156
  • 157
  • 158
  • 159
  • 160
  • 161
  • 162
  • 163
  • 164
  • 165
  • 166
  • 167
  • 168

实际上CheckpointedFunction可以同时应用于keyed state和operator state，测试案例如下：

public class MyFunction<T> implements MapFunction<T, T>, CheckpointedFunction {

    private ReducingState<Long> countPerKey;
    private ListState<Long> countPerPartition;

    private long localCount;

    public void initializeState(FunctionInitializationContext context) throws Exception {
        // get the state data structure for the per-key state
        countPerKey = context.getKeyedStateStore().getReducingState(
                new ReducingStateDescriptor<>("perKeyCount", new AddFunction<>(), Long.class));

        // get the state data structure for the per-partition state
        countPerPartition = context.getOperatorStateStore().getOperatorState(
                new ListStateDescriptor<>("perPartitionCount", Long.class));

        // initialize the "local count variable" based on the operator state
        for (Long l : countPerPartition.get()) {
            localCount += l;
        }
    }

    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // the keyed state is always up to date anyways
        // just bring the per-partition state in shape
        countPerPartition.clear();
        countPerPartition.add(localCount);
    }

    public T map(T value) throws Exception {
        // update the states
        countPerKey.add(1L);
        localCount++;

        return value;
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

下面这个测试案例是数据写出案例：

public class BufferingSink
        implements SinkFunction<Tuple2<String, Integer>>,
                   CheckpointedFunction {
	//用于控制分区
    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, Context contex) throws Exception {
        bufferedElements.add(value);
        //threshold用于控制分区，当数量达到threshold时即作为一个分区写出
        if (bufferedElements.size() == threshold) {
            for (Tuple2<String, Integer> element: bufferedElements) {
                // send it to the sink
            }
            bufferedElements.clear();
        }
    }

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

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Tuple2<String, Integer>> descriptor =
            new ListStateDescriptor<>(
                "buffered-elements",
                TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));

        checkpointedState = context.getOperatorStateStore().getListState(descriptor);
		//当初始化好状态对象后，我们通过 isRestored() 方法判断是否从之前的故障中恢复回来
		//如果该方法返回 true 则表示从故障中进行恢复，会执行接下来的恢复逻辑。
        if (context.isRestored()) {
            for (Tuple2<String, Integer> element : checkpointedState.get()) {
                bufferedElements.add(element);
            }
        }
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

带状态的 Source Function:
带状态的数据源比其他的算子需要注意更多东西。为了保证更新状态以及输出的原子性（用于支持 exactly-once 语义），用户需要在发送数据前获取数据源的全局锁。

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;
    
    /** 存储 state 的变量. */
    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));
            
        // 从我们已保存的状态中恢复 offset 到内存中，在进行任务恢复的时候也会调用此初始化状态的方法
        for (Long l : state.get()) {
            offset = l;
        }
    }

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
2、ListState
3、BroadcastState

详情请看：广播维表

三、Checkpointing

1、用途

flink提供三种开箱即用的State Backend：
1、MemoryStateBackend
MemoryStateBackend内部将状态（state）数据作为对象保存在java堆内存中（taskManager），通过checkpoint机制，MemoryStateBackend将状态（state）进行快照并保存Jobmanager（master）的堆内存中
2、FsStateBackend
FsStateBackend将动态数据保存在taskmanger的内存中，通过checkpoint机制，将状态快照写入配置好的文件系统或目录中。最小元数据保存jobManager的内存中，另外FsStateBackend通过配置一个fileStateThreshold阈值，小于该值时state存储到metadata中而非文件中
3、RocksDBStateBackend
由于RocksDBStateBackend将工作状态存储在taskManger的本地文件系统，状态数量仅仅受限于本地磁盘容量限制，对比于FsStateBackend保存工作状态在内存中，RocksDBStateBackend能避免flink任务持续运行可能导致的状态数量暴增而内存不足的情况，因此适合在生产环境使用

2、RocksDB

1、如果使用java代码进行单任务配置，使用前先加入依赖：

<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
	<version>1.8.0</version>
</dependency>

  • 1
  • 2
  • 3
  • 4
  • 5

2、全局配置，需要在集群Flink客户端conf/flink-conf.yaml中配置：

state.backend: rocksdb
state.backend.incremental: true #启用增量检查点
state.checkpoints.dir:hdfs:///flink-checkpoints # location to store checkpoints

  • 1
  • 2
  • 3

3、任务级别配置状态后端

# 'true' 为是否启用增量检查点
env.setStateBackend(newRocksDBStateBackend("hdfs:///fink-checkpoints", true));

  • 1
  • 2

4、参数调优
Flink RocksDB 状态后端参数调优实践

3、测试案例

四、参考文章

Flink状态管理详解：Keyed State和Operator List State深度解析
flink状态后端配置-设置State Backend
详解Flink的RocksDB状态后端

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/煮酒与君饮/article/detail/1019134
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号