广播状态模式

使用State描述 算子状态，其在恢复时均匀分布在 算子的并行任务中，或者联合，整个状态用于初始化已恢复的并行任务。

第三种支持的算子状态是广播状态。引入广播状态是为了支持一些用例，其中来自一个流的一些数据需要被广播到所有下游任务，在那里它被存储在本地并用于处理另一个流上的所有传入数据元。作为广播状态可以作为自然拟合出现的示例，可以想象包含一组规则的低吞吐量流，我们希望针对来自另一个流的所有数据元进行评估。考虑到上述类型的用例，广播状态与其他算子状态的不同之处在于：

  • 它有一个Map格式，
  • 它仅适用于具有广播流和非__广播流的输入的特定算子，以及
  • 这样的算子可以具有不同名称的多个广播状态。

提供API

为了显示提供的API，我们将在展示其完整函数之前先举例说明。作为我们的运行示例，我们将使用具有不同颜色和形状的对象流的情况，并且我们想要找到跟随特定图案的相同颜色的对象对，例如矩形后跟三角形。我们假设这组有趣的模式随着时间的推移而发展。

在此示例中，第一个流将包含Item具有a Color和Shapeproperty属性的数据元。另一个流将包含Rules。

从流开始Items，我们只需要键入它的Color，因为我们要对相同颜色的。这将确保相同颜色的数据元最终在同一台物理机器上。

  1. // key the shapes by color
  2. KeyedStream<Item, Color> colorPartitionedStream = shapeStream
  3.                         .keyBy(new KeySelector<Shape, Color>(){...});

继续前进Rules，包含它们的流应该被广播到所有下游任务，并且这些任务应该在本地存储它们，以便它们可以针对所有传入来评估它们Items。下面的片段将i）广播规则流和ii）使用提供的MapStateDescriptor，它将创建将存储规则的广播状态。

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

最后，为了评估Rules来自Item流的传入数据元，我们需要：

  • 连接两个流，和

  • 指定我们的匹配检测逻辑。将流（被Keys化或非被Keys化）与a连接BroadcastStream可以通过调用connect()非广播流来完成，并将其BroadcastStream作为参数。这将返回一个BroadcastConnectedStream，我们可以process()使用特殊类型调用CoProcessFunction。该函数将包含我们的匹配逻辑。函数的确切类型取决于非广播流的类型：

  • 如果键入，则函数为a KeyedBroadcastProcessFunction。

  • 如果它是非被Keys化的，则函数是a BroadcastProcessFunction。鉴于我们的非广播流是被Keys化的，以下代码段包含上述调用：

注意：应该在非广播流上调用connect，并将BroadcastStream作为参数。

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

BroadcastProcessFunction和KeyedBroadcastProcessFunction

与a的情况一样CoProcessFunction，这些函数有两种实现方法; 在processBroadcastElement()它负责处理广播流在进入元件和processElement()其用于非广播的一个。这些方法的完整签名如下：

  1. public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {
  3.     public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
  5.     public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
  6. }
  1. public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {
  3.     public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
  5.     public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
  7.     public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
  8. }

首先要注意的是，两个函数都需要实现processBroadcastElement()用于处理广播侧processElement()数据元和非广播侧数据元的方法。

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

这两个上下文（ctx在下面的枚举中）：

  • 允许访问广播状态： ctx.getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)
  • 允许查询数据元的时间戳：ctx.timestamp()，
  • 获取当前水印： ctx.currentWatermark()
  • 得到当前的处理时间：ctx.currentProcessingTime()和
  • 向旁路输出发射数据元：ctx.output(OutputTag<X> outputTag, X value)。的stateDescriptor在getBroadcastState()应该是相同的所述一个在.broadcast(ruleStateDescriptor)上方。

不同之处在于每个人对广播状态的访问类型。广播方对其具有读写访问权限，而非广播方具有只读访问权（因此名称）。原因是在Flink中没有跨任务通信。因此，为了保证广播状态中的内容在我们的 算子的所有并行实例中是相同的，我们只对广播端提供读写访问，广播端在所有任务中看到相同的数据元，并且我们需要对每个任务进行计算。这一侧的传入数据元在所有任务中都是相同的。忽略此规则会破坏状态的一致性保证，从而导致不一致且通常难以调试的结果。

注意： processBroadcast（）中实现的逻辑必须在所有并行实例中具有相同的确定性行为！

最后，由于KeyedBroadcastProcessFunction 算子操作在被Key化的数据流上的事实，它暴露了一些不可用的函数BroadcastProcessFunction。那是：

  • 所述ReadOnlyContext的processElement()方法可以访问Flink的底层定时器服务，其允许注册事件和/或处理时间的定时器。当计时器触发时，onTimer()调用（如上所示），OnTimerContext其中显示与ReadOnlyContextplus相同的函数
    • 能够询问被解雇的计时器是一个事件还是一个处理时间
    • 查询与计时器关联的Keys。
  • 所述Context的processBroadcastElement()方法包含方法applyToKeyedState(StateDescriptor<S, VS> stateDescriptor, KeyedStateFunction<KS, S> function)。这允许注册a KeyedStateFunction以应用于与所提供的所有键相关联的所有状态stateDescriptor。注意：只能在KeyedBroadcastProcessFunction的processElement（）中注册定时器。在processBroadcastElement（）方法中是不可能的，因为没有与广播数据元相关联的键。

回到我们原来的例子，我们KeyedBroadcastProcessFunction可能看起来如下：

  1. new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
  3.     // store partial matches, i.e. first elements of the pair waiting for their second element
  4.     // we keep a list as we may have many first elements waiting
  5.     private final MapStateDescriptor<String, List<Item>> mapStateDesc =
  6.         new MapStateDescriptor<>(
  7.             "items",
  8.             BasicTypeInfo.STRING_TYPE_INFO,
  9.             new ListTypeInfo<>(Item.class));
  11.     // identical to our ruleStateDescriptor above
  12.     private final MapStateDescriptor<String, Rule> ruleStateDescriptor =
  13.         new MapStateDescriptor<>(
  14.             "RulesBroadcastState",
  15.             BasicTypeInfo.STRING_TYPE_INFO,
  16.             TypeInformation.of(new TypeHint<Rule>() {}));
  18.     @Override
  19.     public void processBroadcastElement(Rule value,
  20.                                         Context ctx,
  21.                                         Collector<String> out) throws Exception {
  22.         ctx.getBroadcastState(ruleStateDescriptor).put(value.name, value);
  23.     }
  25.     @Override
  26.     public void processElement(Item value,
  27.                                ReadOnlyContext ctx,
  28.                                Collector<String> out) throws Exception {
  30.         final MapState<String, List<Item>> state = getRuntimeContext().getMapState(mapStateDesc);
  31.         final Shape shape = value.getShape();
  32.     
  33.         for (Map.Entry<String, Rule> entry :
  34.                 ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {
  35.             final String ruleName = entry.getKey();
  36.             final Rule rule = entry.getValue();
  37.     
  38.             List<Item> stored = state.get(ruleName);
  39.             if (stored == null) {
  40.                 stored = new ArrayList<>();
  41.             }
  42.     
  43.             if (shape == rule.second && !stored.isEmpty()) {
  44.                 for (Item i : stored) {
  45.                     out.collect("MATCH: " + i + " - " + value);
  46.                 }
  47.                 stored.clear();
  48.             }
  49.     
  50.             // there is no else{} to cover if rule.first == rule.second
  51.             if (shape.equals(rule.first)) {
  52.                 stored.add(value);
  53.             }
  54.     
  55.             if (stored.isEmpty()) {
  56.                 state.remove(ruleName);
  57.             } else {
  58.                 state.put(ruleName, stored);
  59.             }
  60.         }
  61.     }
  62. }

重要考虑因素

在描述提供的API之后，本节重点介绍使用广播状态时要记住的重要事项。这些是：

  • 没有跨任务通信：如前所述，这就是为什么只有广播方(Keyed)-BroadcastProcessFunction可以修改广播状态的内容的原因。此外，用户必须确保所有任务以相同的方式为每个传入数据元修改广播状态的内容。否则，不同的任务可能具有不同的内容，从而导致不一致的结果。

  • 广播状态中的事件顺序可能因任务而异：尽管广播流的数据元保证所有数据元将（最终）转到所有下游任务，但数据元可能以不同的顺序到达每个任务。因此，每个传入数据元的状态更新不得取决于传入事件的顺序。

  • 所有任务都检查它们的广播状态：虽然检查点发生时所有任务在广播状态中具有相同的数据元（检查点障碍不会覆盖数据元），但所有任务都检查它们的广播状态，而不仅仅是其中一个。这是一个设计决策，以避免在恢复期间从同一文件中读取所有任务（从而避免热点），尽管它的代价是将检查点状态的大小增加p（=并行度）。Flink保证在恢复/重新缩放时不会有重复**数据，也不会丢失数据**。在具有相同或更小并行度的恢复的情况下，每个任务读取其检查点状态。在按比例放大时，每个任务都会读取自己的状态和剩余的任务（p_new-p_old）以循环方式读取先前任务的检查点。

  • 没有RocksDB状态后台：广播状态在运行时保存在内存中，并且应该相应地进行内存配置。这适用于所有算子状态。