Streaming -- State & Fault Tolerance -- The Broadcast State Pattern

Working with State描述了操作符的状态，在恢复时，该状态是均匀地分布在操作符的并行任务之间，或者是统一的，整个状态用于初始化恢复的并行任务。

支持的第三种操作符状态是广播状态。广播状态的引入是为了支持这样的用例:来自一个流的一些数据需要广播到所有下游任务，这些数据在本地存储，用于处理另一个流上的所有传入元素。例如，广播状态可以自然地出现，可以想象一个包含一组规则的低吞吐量流，我们希望根据来自另一个流的所有元素对这些规则进行评估。考虑到上述类型的用例，broadcast状态不同于opera的其他部分

它有一个map格式
它只对具有广播流和非广播流输入的特定操作员可用
这样的操作符可以具有不同名称的多个广播状态。

Provided APIs

为了展示所提供的api，在展示它们的完整功能之前，我们将从一个示例开始。作为我们的运行示例，我们将使用这样的情况，我们有一个不同颜色和形状的对象流，我们希望找到相同颜色的对象对，遵循一定的模式，例如一个矩形后跟一个三角形。我们假设这组有趣的模式会随着时间而演变。

在本例中，第一个流将包含具有颜色和形状属性的Item类型的元素。另一个流将包含规则。

从项目流开始，我们只需要按颜色为其设置键，因为我们需要相同的颜色对。这将确保相同颜色的元素最终出现在相同的物理机器上。

// 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()，以BroadcastStream作为参数，可以连接一个流(键控或非键控)和一个BroadcastStream。这将返回一个BroadcastConnectedStream，我们可以使用特殊类型的CoProcessFunction调用process()。该函数将包含匹配逻辑。该函数的确切类型取决于非广播流的类型:

如果那是键控，然后函数是KeyedBroadcastProcessFunction。
如果它是非键控的，函数是BroadcastProcessFunction。

假设我们的非广播流是键控的，下面的片段包括上面的调用

注意:连接应该在非广播流上调用，以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 and 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。输出(OutputTag<X> OutputTag, X值)。

getBroadcastState()中的stateDescriptor应该与上面.broadcast(ruleStateDescriptor)中的相同。

不同之处在于它们为广播状态提供的访问类型。广播方对它有读写访问权，而非广播方对它有只读访问权(因此名字)。这样做的原因是在Flink中没有跨任务通信。所以,为了保证内容的播出状态是相同的所有平行的实例我们的运营商,我们给广播方面,读写访问只看到相同的元素在所有任务,我们需要计算每个传入的元素,是相同的所有任务。忽略此规则将破坏状态的一致性保证，导致结果不一致，并且通常难以调试结果。

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

最后，由于KeyedBroadcastProcessFunction在键控流上操作，它公开了BroadcastProcessFunction不能使用的一些功能。这是

processElement（）方法中的ReadOnlyContext可以访问Flink的基础计时器服务，该服务可以注册事件和/或处理时间计时器。触发计时器时，将使用OnTimerContext调用onTimer（）（如上所示），该函数提供与ReadOnlyContext plus相同的功能
- 能够询问触发的计时器是事件还是处理时间为1和
- 查询与计时器关联的键。
processBroadcastElement()方法中的上下文包含方法applyToKeyedState(StateDescriptor<S, VS> StateDescriptor, KeyedStateFunction<KS, S>函数)。这允许注册一个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);
            }
        }
    }
}

Important Considerations

在描述了提供的api之后，本节将重点介绍在使用广播状态时需要记住的重要事情。这些都是

没有跨任务通信:如前所述，这就是为什么只有(键控的)-BroadcastProcessFunction的广播端可以修改广播状态的内容的原因。此外，用户必须确保所有任务都以相同的方式为每个传入元素修改广播状态的内容。否则，不同的任务可能具有不同的内容，从而导致结果不一致。
广播状态下的事件顺序可能因任务而异:尽管广播流元素保证了所有元素(最终)将进入所有下游任务，但元素可能以不同的顺序到达每个任务。因此，每个传入元素的状态更新必须不依赖于传入事件的顺序。
所有任务都会检查其广播状态：尽管发生检查点时，所有任务在其广播状态中具有相同的元素（检查点屏障不会越过元素），但所有任务都将指向其广播状态，而不仅仅是其中一个。这是一项设计决策，要避免在还原过程中从同一文件读取所有任务（从而避免出现热点），尽管这样做的代价是将检查点状态的大小增加了p倍（=并行度）。 Flink保证在还原/缩放后不会重复，也不会丢失数据。在使用相同或更小的并行度进行恢复的情况下，每个任务都会读取其检查点状态。扩展后，每个任务将读取其自己的状态，其余任务（p_new-p_old）以循环方式读取先前任务的检查点。
没有RocksDB状态后端:广播状态在运行时保存在内存中，应该相应地进行内存配置。这适用于所有操作符状态。