Flink

广播状态模式

Sat Aug 22, 2020

The Broadcast State Pattern

在本节中，您将了解如何在实践中使用广播状态。请参考 Stateful Stream Processing 来了解有状态流处理背后的概念。

提供的 API #

为了展示所提供的 API，我们将在介绍它们的全部功能之前先举一个例子。作为我们的运行示例，我们将使用这样的情况：我们有一个不同颜色和形状的对象流，我们希望找到相同颜色的对象对，并遵循特定的模式，例如，一个矩形和一个三角形。我们假设有趣的模式集会随着时间的推移而演变。

在这个例子中，第一个流将包含具有 Color 和 Shape 属性的 Item 类型的元素。另一个流将包含 Rules。

从 Items 流开始，我们只需要按 Color keyBy，因为我们想要相同颜色的对。这将确保相同颜色的元素最终会出现在同一个物理机上。

// key the items by color
KeyedStream<Item, Color> colorPartitionedStream = itemStream
                        .keyBy(new KeySelector<Item, Color>(){...});

继续讨论规则，包含规则的流应该被广播到所有下游任务，这些任务应该将它们存储在本地，以便它们可以根据所有传入的项目评估它们。下面的代码段将i)广播规则流，ii)使用提供的 MapStateDescriptor，它将创建规则将被存储的广播状态。

// a map descriptor to store the name of the rule (string) and the rule itself.
MapStateDescriptor<String, Rule> ruleStateDescriptor = new MapStateDescriptor<>(
			"RulesBroadcastState",
			BasicTypeInfo.STRING_TYPE_INFO,
			TypeInformation.of(new TypeHint<Rule>() {}));
		
// broadcast the rules and create the broadcast state
BroadcastStream<Rule> ruleBroadcastStream = ruleStream
                        .broadcast(ruleStateDescriptor);

最后，为了根据从 Item 流传入的元素来评估 Rules，我们需要。

连接(connect)两个流，并且
指定我们的匹配检测逻辑。

将一个流（keyed or non-keyed）与 BroadcastStream 连接起来，可以通过在非广播流上调用 connect() 来完成，并将 BroadcastStream 作为一个参数。这将返回一个 BroadcastConnectedStream，我们可以在这个 Stream 上调用一个特殊类型的 CoProcessFunction 来处理。该函数将包含我们的匹配逻辑。该函数的具体类型取决于非广播流的类型。

如果它是 keyed，那么这个函数就是 KeyedBroadcastProcessFunction。
如果是 non-keyed,，那么该函数就是一个 BroadcastProcessFunction。

鉴于我们的非广播流是 keyed 的，下面的代码段包含了上述调用。

注意：连接(connect)应该被调用在非广播流上，以 BroadcastStream 作为参数。

DataStream<String> output = colorPartitionedStream
                 .connect(ruleBroadcastStream)
                 .process(
                     
                     // type arguments in our KeyedBroadcastProcessFunction represent: 
                     //   1. the key of the keyed stream
                     //   2. the type of elements in the non-broadcast side
                     //   3. the type of elements in the broadcast side
                     //   4. the type of the result, here a string
                     
                     new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
                         // my matching logic
                     }
                 );

BroadcastProcessFunction 和 KeyedBroadcastProcessFunction #

与 CoProcessFunction 一样，这些函数有两个处理方法要实现；processBroadcastElement() 负责处理广播流中的传入元素，processElement() 用于处理非广播流。这些方法的完整签名如下:

public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {

    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;

    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}
public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {

    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;

    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;

    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}

首先需要注意的是，这两个函数在处理广播端元素时都需要实现 processBroadcastElement() 方法，在处理非广播端元素时需要实现 processElement() 方法。

这两个方法在提供的上下文中有所不同。非广播侧有一个 ReadOnlyContext，而广播侧有一个 Context。

这两个上下文（以下枚举中的 ctx）:

提供对广播状态的访问：ctx.getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)。
允许查询元素的时间戳：ctx.timestamp()。
获取当前水印：ctx.currentWatermark()。
获取当前处理时间：ctx.currentProcessingTime()，以及
将元素发射到侧输出：ctx.output(OutputTag<X> outputTag, X value)。

getBroadcastState() 中的 stateDescriptor 应该和上面的 .broadcast(ruleStateDescriptor) 中的 stateDescriptor 是一样的。

区别在于各自对广播状态的访问类型。广播端对其有读写访问权，而非广播端则只有读的访问权（因此才有这些名字）。原因是在 Flink 中，不存在跨任务通信。所以，为了保证广播状态中的内容在我们操作符的所有并行实例中都是相同的，我们只给广播侧读写访问权，而广播侧在所有任务中看到的元素都是相同的，并且我们要求该侧每个传入元素的计算在所有任务中都是相同的。忽略这个规则会打破状态的一致性保证，导致结果不一致，而且往往难以调试。

注意 processBroadcastElement() 中实现的逻辑必须在所有并行实例中具有相同的确定性行为!

最后，由于 KeyedBroadcastProcessFunction 是在 keyed stream 上运行的，它暴露了一些 BroadcastProcessFunction 无法实现的功能。那就是

processElement() 方法中的 ReadOnlyContext 允许访问 Flink 的底层定时器服务，它允许注册事件和/或处理时间定时器。当一个定时器发射时， onTimer() (如上所示)被调用一个 OnTimerContext，它暴露了与 ReadOnlyContext 相同的功能，再加上

能够询问发射的定时器是事件还是处理时间, 和
来查询与定时器相关联的键。

processBroadcastElement() 方法中的 Context 包含 applyToKeyedState(StateDescriptor<S, VS> stateDescriptor, KeyedStateFunction<KS, S> function) 方法。这允许注册一个 KeyedStateFunction，以应用于与提供的 stateDescriptor 相关联的所有键的所有状态。

注意。注册定时器只能在 KeyedBroadcastProcessFunction 的 processElement() 处进行，而且只能在那里进行。在 processBroadcastElement() 方法中是不可能的，因为没有键与广播元素相关联。

回到我们原来的例子，我们的 KeyedBroadcastProcessFunction 可以是如下的样子。

new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {

    // store partial matches, i.e. first elements of the pair waiting for their second element
    // we keep a list as we may have many first elements waiting
    private final MapStateDescriptor<String, List<Item>> mapStateDesc =
        new MapStateDescriptor<>(
            "items",
            BasicTypeInfo.STRING_TYPE_INFO,
            new ListTypeInfo<>(Item.class));

    // identical to our ruleStateDescriptor above
    private final MapStateDescriptor<String, Rule> ruleStateDescriptor = 
        new MapStateDescriptor<>(
            "RulesBroadcastState",
            BasicTypeInfo.STRING_TYPE_INFO,
            TypeInformation.of(new TypeHint<Rule>() {}));

    @Override
    public void processBroadcastElement(Rule value,
                                        Context ctx,
                                        Collector<String> out) throws Exception {
        ctx.getBroadcastState(ruleStateDescriptor).put(value.name, value);
    }

    @Override
    public void processElement(Item value,
                               ReadOnlyContext ctx,
                               Collector<String> out) throws Exception {

        final MapState<String, List<Item>> state = getRuntimeContext().getMapState(mapStateDesc);
        final Shape shape = value.getShape();
    
        for (Map.Entry<String, Rule> entry :
                ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {
            final String ruleName = entry.getKey();
            final Rule rule = entry.getValue();
    
            List<Item> stored = state.get(ruleName);
            if (stored == null) {
                stored = new ArrayList<>();
            }
    
            if (shape == rule.second && !stored.isEmpty()) {
                for (Item i : stored) {
                    out.collect("MATCH: " + i + " - " + value);
                }
                stored.clear();
            }
    
            // there is no else{} to cover if rule.first == rule.second
            if (shape.equals(rule.first)) {
                stored.add(value);
            }
    
            if (stored.isEmpty()) {
                state.remove(ruleName);
            } else {
                state.put(ruleName, stored);
            }
        }
    }
}

重要的考虑因素 #

在介绍完提供的 API 之后，本节重点介绍使用广播状态时需要注意的重要事项。这些事项是

没有跨任务通信。如前所述，这就是为什么只有 (Keyed)-BroadcastProcessFunction 的广播端可以修改广播状态的内容的原因。此外，用户必须确保所有的任务对每一个传入元素都以同样的方式修改广播状态的内容。否则，不同的任务可能有不同的内容，导致结果不一致。
不同任务的广播状态中事件的顺序可能不同。虽然广播流的元素保证了所有元素将（最终）进入所有下游任务，但元素可能会以不同的顺序到达每个任务。因此，每个传入元素的状态更新必须不依赖于传入事件的顺序。
所有的任务都会对其广播状态进行 checkpoint。虽然当 checkpoint 发生时，所有任务的广播状态中都有相同的元素（checkpoint 屏障不会超过元素），但所有任务都会 checkpoint 他们的广播状态，而不仅仅是其中一个。这是一个设计决定，以避免在还原过程中让所有任务从同一个文件中读取（从而避免热点），尽管它的代价是将检查点状态的大小增加了p的系数（=并行性）。Flink 保证在恢复/缩放时，不会有重复和丢失的数据。在以相同或更小的并行度进行恢复时，每个任务读取其检查点状态。扩容后，每个任务读取自己的状态，其余任务（p_new-p_old）以循环的方式读取之前任务的检查点。
没有 RocksDB 状态后端。广播状态在运行时保存在内存中，内存供应也应相应进行。这对所有的操作符状态都适用。

原文链接: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/broadcast_state.html