undraw_annotation_7das.png

Why : 为什么会出现

举一个例子：
Flink消费Kafka的数据后 与Mysql的规则数据做匹配<动态维护>,当Operator中的Task消费到Kafka的数据后，看到有查询Mysql的操作就都会去查询数据库。注意这里的查询可能有一个可能有多个查询 最终一个节点中可能会存在多份Mysql的数据, 引发的问题: 对Mysql是一个负载的过程,另一方面Shuffer过程比较消耗资源<网络Io-PRC请求>。 针对此问题: 想到Spark的广播—> Driver端先去数据库拿数据后广播到Worker端,这样的话直接从本地读取就会快很多。

从1.5.0版开始，Apache Flink具有一种新的状态类型，称为广播状态。Apache Flink广播状态实用指南

What：是什么

本质上是一个状态 —> Map类型的State 广播状态可用于以特定方式合并和联合处理两个事件流。第一个流的事件会广播到运算符的所有并行实例，该实例将它们保持为状态。另一个流的事件不广播，而是发送到同一运算符的各个实例，并与广播流的事件一起处理。对于需要加入低吞吐量和高吞吐量流或需要动态更新其处理逻辑的应用程序，新的广播状态非常合适。

广播状态（Broadcast State）的引入是为了支持一些来自一个流的数据需要广播到所有下游任务的情况，它存储在本地，用于处理其他流上的所有传入元素。

考虑到上述类型的情况，broadcast状态与其他运算符状态的不同之处在于:

  • 它有一种Map格式。
  • 它仅对具有广播流和非广播流作为输入的特定操作符可用。
  • 并且这样的操作符可以有不同名称的多个广播状态。


Where：什么情况？组成部分

  1. 需要有两个数据流 DataStream 与 BroadcastStream
  2. MapStateDescriptor 广播的类型
  3. connect 后 process 后传入特定的CoProcessFunction
  4. 分别获取broadCast后 广播方调用上下文对象put近State中，非广播方获取State。collect(Return 的值)
  1.  //可以有多种类型 以及类型的种类 基本数据类型 Pojo List ...
  2.  MapStateDescriptor<String, String> config = new MapStateDescriptor("config", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
  3.  MapStateDescriptor<Void, UserInfoModel> config1 = new MapStateDescriptor("config", BasicTypeInfo.VOID_TYPE_INFO, TypeInformation.of(UserInfoModel.class));
  4.  MapStateDescriptor<Void, List<UserInfoModel>> config2 = new MapStateDescriptor("config", BasicTypeInfo.VOID_TYPE_INFO, new ListTypeInfo(UserInfoModel.class));

How：如何操作

  1. DataStream<String> output = colorPartitionedStream
  2.                  .connect(ruleBroadcastStream)
  3.                  .process(
  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
  11.                      new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
  12.                          // my matching logic
  13.                      }
  15.                     new BroadcastProcessFunction<Color, Item, Rule, String>() {
  16.                          // my matching logic
  17.                     }
  18.                  );

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

BroadcastProcessFunction

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

KeyedBroadcastProcessFunction

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

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

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

这两种情况(以下列举的ctx)：

访问广播状态:ctx.getBroadcastState(MapStateDescriptor stateDescriptor)。

允许查询元素的时间戳:ctx.timestamp()。

获取当前水印:ctx.currentWatermark()。

获取当前处理时间:ctx.currentProcessingTime()。

向侧输出发送元素:ctx.output(OutputTag outputTag, X value)。

getBroadcastState()中的状态描述符(stateDescriptor)应该与.broadcast(ruleStateDescriptor)中的状态描述符相同。

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

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

最后，由于KeyedBroadcastProcessFunction是在一个键流上操作的，它公开了一些BroadcastProcessFunction不可用的功能。那就是:

  1. 所述ReadOnlyContext的processElement()方法可以访问Flink的底层定时器服务，其允许注册事件和/或处理时间的定时器。当计时器触发时，onTimer()会使用调用（如上所示），OnTimerContext该公开了与ReadOnlyContextplus相同的功能。
    • 询问触发的计时器是事件还是处理时间的能力，并且
    • 查询与计时器关联的键。
  2. 所述Context的processBroadcastElement()方法包含方法 applyToKeyedState(StateDescriptor<S, VS> stateDescriptor, KeyedStateFunction<KS, S> function)。这允许一个注册KeyedStateFunction将被施加到所有键的所有状态与所提供的相关联stateDescriptor。

注意：仅在“ KeyedBroadcastProcessFunction”的“ processElement（）”处才可以注册计时器。在processBroadcastElement（）方法中是不可能的，因为没有与广播元素关联的键。

new BroadcastProcessFunction

  1.  /**
  2.   * KafkaPojo: KafkaPojo(name=张三, age=18, sex=1, timeStamp=1576300491032)
  3.   * String : 张三##18##1##1576300491032
  4.   */
  5. .process(new BroadcastProcessFunction<KafkaPojo, String, String>() {
  7.             String strConfig = "config";
  9.             @Override
  10.             public void processElement(KafkaPojo kafkaPojo, ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {
  12.                 ReadOnlyBroadcastState<String, String> state = readOnlyContext.getBroadcastState(config);
  14.                 int userAge = Integer.parseInt(state.get(strConfig).split("##")[1]);
  16.                 if (userAge == kafkaPojo.getAge()) {
  17.                     collector.collect("Good");
  18.                 }else{
  19.                     collector.collect("No Good");
  20.                 }
  22.             }
  24.             @Override
  25.             public void processBroadcastElement(String str, Context context, Collector<String> collector) throws Exception {
  26.                 BroadcastState<String, String> state = context.getBroadcastState(config);
  27.                 state.clear();
  28.                 state.put(strConfig, str);
  30.             }
  31.         }).setParallelism(1).print();

new 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();
  33.         for (Map.Entry<String, Rule> entry :
  34.                 ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {
  35.             final String ruleName = entry.getKey();
  36.             final Rule rule = entry.getValue();
  38.             List<Item> stored = state.get(ruleName);
  39.             if (stored == null) {
  40.                 stored = new ArrayList<>();
  41.             }
  43.             if (shape == rule.second && !stored.isEmpty()) {
  44.                 for (Item i : stored) {
  45.                     out.collect("MATCH: " + i + " - " + value);
  46.                 }
  47.                 stored.clear();
  48.             }
  50.             // there is no else{} to cover if rule.first == rule.second
  51.             if (shape.equals(rule.first)) {
  52.                 stored.add(value);
  53.             }
  55.             if (stored.isEmpty()) {
  56.                 state.remove(ruleName);
  57.             } else {
  58.                 state.put(ruleName, stored);
  59.             }
  60.         }
  61.     }
  62. }

使用时的注意事项：

使用broadcast state时需要记住的重要内容。

  • 没有跨任务通信（There is no cross-task communication）： 如前所述，这就是为什么只有(键的)broadcastprocessfunction的广播端可以修改广播状态的内容。此外，用户必须确保所有任务以相同的方式为每个传入元素修改broadcast状态的内容。
    否则，不同的任务可能具有不同的内容，从而导致不一致的结果。
  • 广播状态下的事件顺序可能会因任务的不同而不同（Order of events in Broadcast State may differ across tasks）：尽管广播流的元素可以保证所有元素(最终)都将进入所有下游任务，元素可能以不同的顺序到达每个任务。因此，每个传入元素的状态更新不能依赖于传入事件的顺序。
  • 所有任务检查它们的广播状态（All tasks checkpoint their broadcast state）：尽管在检查点发生时，所有任务的广播状态中都有相同的元素(检查点屏障不会跨越元素)，所有任务检查它们的广播状态，而且不止一个。这是一个设计决策，避免在恢复期间从同一个文件读取所有任务(从而避免热点)，尽管这样做的代价是将检查点状态的大小增加p(=并行度)。Flink保证在恢复/重新扫描时不会有重复和丢失数据。在并行度相同或更小的恢复情况下，每个任务读取其检查点状态。在扩展时，每个任务读取自己的状态，其余任务(p_new-p_old)以循环方式读取以前任务的检查点。
  • 没有RocksDB状态后端（No RocksDB state backend）：广播状态在运行时保存在内存中，应该相应地进行内存供应。这适用于所有的运算符状态。

注意ConnectedStream 与 Union的区别

• 连接的流的数量 连接类型 处理方法的数量
• 1. ConnectedStreamed只能连接两个流，而union可以连接多于两个流。
• 2. ConnectedStream连接的两个流类型可以不一致，而union连接的流类型必须是一致的。
• 3. ConnectedStreams会对两个流的数据应用不同的处理方法，并且双流之间可以共享状态。