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

【Flink 源码系列】Flink 源码：广播流状态源码解析_flink广播流

作者：繁依Fanyi0 | 2024-06-01 23:30:59

赞

踩

flink广播流


2ed91fd8a6b201875a9116858b27be3b.gif

Broadcast State 是 Operator State 的一种特殊类型。它的引入是为了支持这样的场景: 一个流的记录需要广播到所有下游任务，在这些用例中，它们用于在所有子任务中维护相同的状态。然后可以在处理第二个流的数据时访问这个广播状态，广播状态有自己的一些特性。

  1. 必须定义为一个 Map 结构。

  2. 广播状态只能在广播流侧修改，非广播侧不能修改状态。

  3. Broadcast State 运行时的状态只能保存在内存中。

看到这相信你肯定会有下面的疑问：

  • 广播状态为什么必须定义为 Map 结构，我用其他的状态类型不行吗？

  • 广播状态为什么只能在广播侧修改，非广播侧为什么不能修改呢？

  • 广播状态为什么只能保存在内存中，难道不能用 Rockdb 状态后端吗？

下面就带着这三个疑问通过阅读相关源码，回答上面的问题。

broadcast 源码

  1. /**
  2.  * Sets the partitioning of the {@link DataStream} so that the output elements are broadcasted
  3.  * to every parallel instance of the next operation. In addition, it implicitly as many {@link
  4.  * org.apache.flink.api.common.state.BroadcastState broadcast states} as the specified
  5.  * descriptors which can be used to store the element of the stream.
  6.  *
  7.  * @param broadcastStateDescriptors the descriptors of the broadcast states to create.
  8.  * @return A {@link BroadcastStream} which can be used in the {@link #connect(BroadcastStream)}
  9.  *     to create a {@link BroadcastConnectedStream} for further processing of the elements.
  10.  */
  11. @PublicEvolving
  12. public BroadcastStream<T> broadcast(
  13.         final MapStateDescriptor<?, ?>... broadcastStateDescriptors) {
  14.     Preconditions.checkNotNull(broadcastStateDescriptors);
  15.     final DataStream<T> broadcastStream = setConnectionType(new BroadcastPartitioner<>());
  16.     return new BroadcastStream<>(environment, broadcastStream, broadcastStateDescriptors);
  17. }

可以发现 broadcast 方法需要的参数是 MapStateDescriptor 也就是一个 Map 结构的状态描述符，我们在使用的时候就必须定义为 MapStateDescriptor，否则会直接报错，其实主要是因为广播状态的作用是和非广播流进行关联，你可以想象成双流 join 的场景，那么 join 的时候就必须要有一个主键，也就是相同的 key 才能 join 上，所以 Map（key-value） 结构是最适合这种场景的，key 可以存储要关联字段，value 可以是任意类型的广播数据，在关联的时候只需要获取到广播状态，然后 state.get(key) 就可以很容易拿到广播数据。

process 源码

  1. @PublicEvolving
  2. public <KEY, OUT> SingleOutputStreamOperator<OUT> process(
  3.         final KeyedBroadcastProcessFunction<KEY, IN1, IN2, OUT> function) {
  4.   // 获取输出数据的类型信息
  5.     TypeInformation<OUT> outTypeInfo =
  6.             TypeExtractor.getBinaryOperatorReturnType(
  7.                     function,
  8.                     KeyedBroadcastProcessFunction.class,
  9.                     1,
  10.                     2,
  11.                     3,
  12.                     TypeExtractor.NO_INDEX,
  13.                     getType1(),
  14.                     getType2(),
  15.                     Utils.getCallLocationName(),
  16.                     true);
  17.  
  18.     return process(function, outTypeInfo);
  19. }

process 方法需要的参数是 KeyedBroadcastProcessFunction<KEY, IN1, IN2, OUT>，跟普通的 KeyedProcessFunction<K, I, O> 相比，很容易发现多了一个泛型参数，因为这里的 process 上游连接的是两个数据流，所以需要两个类型。然后调用 process 的重载方法。

process 源码

  1. @PublicEvolving
  2. public <KEY, OUT> SingleOutputStreamOperator<OUT> process(
  3.         final KeyedBroadcastProcessFunction<KEY, IN1, IN2, OUT> function,
  4.         final TypeInformation<OUT> outTypeInfo) {
  5.  
  6.     Preconditions.checkNotNull(function);
  7.     Preconditions.checkArgument(
  8.             nonBroadcastStream instanceof KeyedStream,
  9.             "A KeyedBroadcastProcessFunction can only be used on a keyed stream.");
  10.   
  11.     return transform(function, outTypeInfo);
  12. }

这个 process 方法里面什么都没干，直接调用 transform 方法。

transform 源码

  1. @Internal
  2. private <KEY, OUT> SingleOutputStreamOperator<OUT> transform(
  3.         final KeyedBroadcastProcessFunction<KEY, IN1, IN2, OUT> userFunction,
  4.         final TypeInformation<OUT> outTypeInfo) {
  5.  
  6.     // read the output type of the input Transforms to coax out errors about MissingTypeInfo
  7.     nonBroadcastStream.getType();
  8.     broadcastStream.getType();
  9.  
  10.     KeyedStream<IN1, KEY> keyedInputStream = (KeyedStream<IN1, KEY>) nonBroadcastStream;
  11.   // 构造 KeyedBroadcastStateTransformation
  12.     final KeyedBroadcastStateTransformation<KEY, IN1, IN2, OUT> transformation =
  13.             new KeyedBroadcastStateTransformation<>(
  14.                     "Co-Process-Broadcast-Keyed",
  15.                     nonBroadcastStream.getTransformation(),
  16.                     broadcastStream.getTransformation(),
  17.                     clean(userFunction),
  18.                     broadcastStateDescriptors,
  19.                     keyedInputStream.getKeyType(),
  20.                     keyedInputStream.getKeySelector(),
  21.                     outTypeInfo,
  22.                     environment.getParallelism());
  23.  
  24.     @SuppressWarnings({"unchecked", "rawtypes"})
  25.     final SingleOutputStreamOperator<OUT> returnStream =
  26.             new SingleOutputStreamOperator(environment, transformation);
  27.   // 添加到 List<Transformation<?>> 集合
  28.     getExecutionEnvironment().addOperator(transformation);
  29.     return returnStream;
  30. }

transform 方法里面主要做了两件事：

  1. 先是构造对应的 KeyedBroadcastStateTransformation 对象，其实  KeyedBroadcastStateTransformation 也是 Transformation 的一个子类。

  2. 然后把构造好的 transformation 添加到 List<Transformation<?>> 集合里，后面在构建 StreamGraph 的时候会从这个集合里获取 Transformation。

getStreamGraph 源码

  1. @Internal
  2. public StreamGraph getStreamGraph(boolean clearTransformations) {
  3.     final StreamGraph streamGraph = getStreamGraphGenerator(transformations).generate();
  4.     if (clearTransformations) {
  5.         transformations.clear();
  6.     }
  7.     return streamGraph;
  8. }

getStreamGraph 的主要作用就是生成 StreamGraph。下面就会用到上一步生成的 List<Transformation<?>> 集合，因为这篇文章主要是分析 Flink 广播流的源码，所以只会对广播流相关的源码进行解析。

getStreamGraphGenerator 源码

  1. private StreamGraphGenerator getStreamGraphGenerator(List<Transformation<?>> transformations) {
  2.     if (transformations.size() <= 0) {
  3.         throw new IllegalStateException(
  4.                 "No operators defined in streaming topology. Cannot execute.");
  5.     }
  6.  
  7.     // We copy the transformation so that newly added transformations cannot intervene with the
  8.     // stream graph generation.
  9.     return new StreamGraphGenerator(
  10.                     new ArrayList<>(transformations), config, checkpointCfg, configuration)
  11.             .setStateBackend(defaultStateBackend)
  12.             .setChangelogStateBackendEnabled(changelogStateBackendEnabled)
  13.             .setSavepointDir(defaultSavepointDirectory)
  14.             .setChaining(isChainingEnabled)
  15.             .setUserArtifacts(cacheFile)
  16.             .setTimeCharacteristic(timeCharacteristic)
  17.             .setDefaultBufferTimeout(bufferTimeout)
  18.             .setSlotSharingGroupResource(slotSharingGroupResources);
  19. }

getStreamGraphGenerator 方法主要就是构造 StreamGraphGenerator 对象，StreamGraphGenerator 构造完成后，就可以调用 generate 方法来产生 StreamGraph 了，在看 generate 方法之前先来看一下 StreamGraphGenerator 的静态代码块。

StreamGraphGenerator 源码

  1. static {
  2.     @SuppressWarnings("rawtypes")
  3.     Map<Class<? extends Transformation>, TransformationTranslator<?, ? extends Transformation>>
  4.             tmp = new HashMap<>();
  5.     tmp.put(OneInputTransformation.class, new OneInputTransformationTranslator<>());
  6.     tmp.put(TwoInputTransformation.class, new TwoInputTransformationTranslator<>());
  7.     tmp.put(MultipleInputTransformation.class, new MultiInputTransformationTranslator<>());
  8.     tmp.put(KeyedMultipleInputTransformation.class, new MultiInputTransformationTranslator<>());
  9.     tmp.put(SourceTransformation.class, new SourceTransformationTranslator<>());
  10.     tmp.put(SinkTransformation.class, new SinkTransformationTranslator<>());
  11.     tmp.put(LegacySinkTransformation.class, new LegacySinkTransformationTranslator<>());
  12.     tmp.put(LegacySourceTransformation.class, new LegacySourceTransformationTranslator<>());
  13.     tmp.put(UnionTransformation.class, new UnionTransformationTranslator<>());
  14.     tmp.put(PartitionTransformation.class, new PartitionTransformationTranslator<>());
  15.     tmp.put(SideOutputTransformation.class, new SideOutputTransformationTranslator<>());
  16.     tmp.put(ReduceTransformation.class, new ReduceTransformationTranslator<>());
  17.     tmp.put(
  18.             TimestampsAndWatermarksTransformation.class,
  19.             new TimestampsAndWatermarksTransformationTranslator<>());
  20.     tmp.put(BroadcastStateTransformation.class, new BroadcastStateTransformationTranslator<>());
  21.     tmp.put(
  22.             KeyedBroadcastStateTransformation.class,
  23.             new KeyedBroadcastStateTransformationTranslator<>());
  24.     translatorMap = Collections.unmodifiableMap(tmp);
  25. }

在初始化 StreamGraphGenerator 之前，会先执行其静态代码块生成一个 Transformation -> TransformationTranslator 映射关系的 Map 集合，后面会用到这个 Map。

transform 源码

  1. // 根据 Transformation 获取对应的 TransformationTranslator 
  2. final TransformationTranslator<?, Transformation<?>> translator =
  3.         (TransformationTranslator<?, Transformation<?>>)
  4.                 translatorMap.get(transform.getClass());
  5.  
  6. Collection<Integer> transformedIds;
  7. if (translator != null) {
  8.    
  9.     transformedIds = translate(translator, transform);
  10. } else {
  11.     transformedIds = legacyTransform(transform);
  12. }

构造完 StreamGraphGenerator 对象后，紧接着会调用 generate 方法，然后又调用了 transform 方法，这里会从上面生成的 Map 里面获取到对应的 TransformationTranslator，然后调用 translate 方法。

translate#translateForStreaming#translateForStreamingInternal 源码

  1. @Override
  2. protected Collection<Integer> translateForStreamingInternal(
  3.         final KeyedBroadcastStateTransformation<KEY, IN1, IN2, OUT> transformation,
  4.         final Context context) {
  5.     checkNotNull(transformation);
  6.     checkNotNull(context);
  7.   // 构建 CoBroadcastWithKeyedOperator 
  8.     CoBroadcastWithKeyedOperator<KEY, IN1, IN2, OUT> operator =
  9.             new CoBroadcastWithKeyedOperator<>(
  10.                     transformation.getUserFunction(),
  11.                     transformation.getBroadcastStateDescriptors());
  12.  
  13.     return translateInternal(
  14.             transformation,
  15.             transformation.getRegularInput(),
  16.             transformation.getBroadcastInput(),
  17.             SimpleOperatorFactory.of(operator),
  18.             transformation.getStateKeyType(),
  19.             transformation.getKeySelector(),
  20.             null /* no key selector on broadcast input */,
  21.             context);
  22. }

translate 方法最终会调用到 KeyedBroadcastStateTransformationTranslator 的 translateForStreamingInternal 方法中，根据 UserFunction（用户代码）和 broadcastStateDescriptors（广播状态描述符）构造CoBroadcastWithKeyedOperator 对象。

CoBroadcastWithKeyedOperator 源码

  1. /**
  2.  * A {@link TwoInputStreamOperator} for executing {@link KeyedBroadcastProcessFunction
  3.  * KeyedBroadcastProcessFunctions}.
  4.  *
  5.  * @param <KS> The key type of the input keyed stream.
  6.  * @param <IN1> The input type of the keyed (non-broadcast) side.
  7.  * @param <IN2> The input type of the broadcast side.
  8.  * @param <OUT> The output type of the operator.
  9.  */
  10. @Internal
  11. public class CoBroadcastWithKeyedOperator<KS, IN1, IN2, OUT>
  12.         extends AbstractUdfStreamOperator<OUT, KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT>>
  13.         implements TwoInputStreamOperator<IN1, IN2, OUT>, Triggerable<KS, VoidNamespace> {
  14.  
  15.     private static final long serialVersionUID = 5926499536290284870L;
  16.  
  17.     private final List<MapStateDescriptor<?, ?>> broadcastStateDescriptors;
  18.  
  19.     private transient TimestampedCollector<OUT> collector;
  20.  
  21.     private transient Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> broadcastStates;
  22.  
  23.     private transient ReadWriteContextImpl rwContext;
  24.  
  25.     private transient ReadOnlyContextImpl rContext;
  26.  
  27.     private transient OnTimerContextImpl onTimerContext;
  28.  
  29.     public CoBroadcastWithKeyedOperator(
  30.             final KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> function,
  31.             final List<MapStateDescriptor<?, ?>> broadcastStateDescriptors) {
  32.         super(function);
  33.         this.broadcastStateDescriptors = Preconditions.checkNotNull(broadcastStateDescriptors);
  34.     }
  35.  
  36.     @Override
  37.     public void open() throws Exception {
  38.         super.open();
  39.  
  40.         InternalTimerService<VoidNamespace> internalTimerService =
  41.                 getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);
  42.  
  43.         TimerService timerService = new SimpleTimerService(internalTimerService);
  44.  
  45.         collector = new TimestampedCollector<>(output);
  46.  
  47.         this.broadcastStates = new HashMap<>(broadcastStateDescriptors.size());
  48.         for (MapStateDescriptor<?, ?> descriptor : broadcastStateDescriptors) {
  49.             broadcastStates.put(
  50.                     descriptor, 
  51.               // 初始化状态实现实例
  52.               getOperatorStateBackend().getBroadcastState(descriptor));
  53.         }
  54.  
  55.         rwContext =
  56.                 new ReadWriteContextImpl(
  57.                         getExecutionConfig(),
  58.                         getKeyedStateBackend(),
  59.                         userFunction,
  60.                         broadcastStates,
  61.                         timerService);
  62.         rContext =
  63.                 new ReadOnlyContextImpl(
  64.                         getExecutionConfig(), userFunction, broadcastStates, timerService);
  65.         onTimerContext =
  66.                 new OnTimerContextImpl(
  67.                         getExecutionConfig(), userFunction, broadcastStates, timerService);
  68.     }
  69.  
  70.     @Override
  71.     public void processElement1(StreamRecord<IN1> element) throws Exception {
  72.         collector.setTimestamp(element);
  73.         rContext.setElement(element);
  74.         userFunction.processElement(element.getValue(), rContext, collector);
  75.         rContext.setElement(null);
  76.     }
  77.  
  78.     @Override
  79.     public void processElement2(StreamRecord<IN2> element) throws Exception {
  80.         collector.setTimestamp(element);
  81.         rwContext.setElement(element);
  82.         userFunction.processBroadcastElement(element.getValue(), rwContext, collector);
  83.         rwContext.setElement(null);
  84.     }
  85.  
  86.     private class ReadWriteContextImpl
  87.             extends KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT>.Context {
  88.  
  89.         private final ExecutionConfig config;
  90.  
  91.         private final KeyedStateBackend<KS> keyedStateBackend;
  92.  
  93.         private final Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> states;
  94.  
  95.         private final TimerService timerService;
  96.  
  97.         private StreamRecord<IN2> element;
  98.  
  99.         ReadWriteContextImpl(
  100.                 final ExecutionConfig executionConfig,
  101.                 final KeyedStateBackend<KS> keyedStateBackend,
  102.                 final KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> function,
  103.                 final Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> broadcastStates,
  104.                 final TimerService timerService) {
  105.  
  106.             function.super();
  107.             this.config = Preconditions.checkNotNull(executionConfig);
  108.             this.keyedStateBackend = Preconditions.checkNotNull(keyedStateBackend);
  109.             this.states = Preconditions.checkNotNull(broadcastStates);
  110.             this.timerService = Preconditions.checkNotNull(timerService);
  111.         }
  112.  
  113.         void setElement(StreamRecord<IN2> e) {
  114.             this.element = e;
  115.         }
  116.  
  117.         @Override
  118.         public Long timestamp() {
  119.             checkState(element != null);
  120.             return element.getTimestamp();
  121.         }
  122.  
  123.         @Override
  124.         public <K, V> BroadcastState<K, V> getBroadcastState(
  125.                 MapStateDescriptor<K, V> stateDescriptor) {
  126.             Preconditions.checkNotNull(stateDescriptor);
  127.  
  128.             stateDescriptor.initializeSerializerUnlessSet(config);
  129.             BroadcastState<K, V> state = (BroadcastState<K, V>) states.get(stateDescriptor);
  130.             if (state == null) {
  131.                 throw new IllegalArgumentException(
  132.                         "The requested state does not exist. "
  133.                                 + "Check for typos in your state descriptor, or specify the state descriptor "
  134.                                 + "in the datastream.broadcast(...) call if you forgot to register it.");
  135.             }
  136.             return state;
  137.         }
  138.  
  139.         @Override
  140.         public <X> void output(OutputTag<X> outputTag, X value) {
  141.             checkArgument(outputTag != null, "OutputTag must not be null.");
  142.             output.collect(outputTag, new StreamRecord<>(value, element.getTimestamp()));
  143.         }
  144.  
  145.         @Override
  146.         public long currentProcessingTime() {
  147.             return timerService.currentProcessingTime();
  148.         }
  149.  
  150.         @Override
  151.         public long currentWatermark() {
  152.             return timerService.currentWatermark();
  153.         }
  154.  
  155.         @Override
  156.         public <VS, S extends State> void applyToKeyedState(
  157.                 final StateDescriptor<S, VS> stateDescriptor,
  158.                 final KeyedStateFunction<KS, S> function)
  159.                 throws Exception {
  160.  
  161.             keyedStateBackend.applyToAllKeys(
  162.                     VoidNamespace.INSTANCE,
  163.                     VoidNamespaceSerializer.INSTANCE,
  164.                     Preconditions.checkNotNull(stateDescriptor),
  165.                     Preconditions.checkNotNull(function));
  166.         }
  167.     }
  168.  
  169.     private class ReadOnlyContextImpl extends ReadOnlyContext {
  170.  
  171.         private final ExecutionConfig config;
  172.  
  173.         private final Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> states;
  174.  
  175.         private final TimerService timerService;
  176.  
  177.         private StreamRecord<IN1> element;
  178.  
  179.         ReadOnlyContextImpl(
  180.                 final ExecutionConfig executionConfig,
  181.                 final KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> function,
  182.                 final Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> broadcastStates,
  183.                 final TimerService timerService) {
  184.  
  185.             function.super();
  186.             this.config = Preconditions.checkNotNull(executionConfig);
  187.             this.states = Preconditions.checkNotNull(broadcastStates);
  188.             this.timerService = Preconditions.checkNotNull(timerService);
  189.         }
  190.  
  191.         void setElement(StreamRecord<IN1> e) {
  192.             this.element = e;
  193.         }
  194.  
  195.         @Override
  196.         public Long timestamp() {
  197.             checkState(element != null);
  198.             return element.hasTimestamp() ? element.getTimestamp() : null;
  199.         }
  200.  
  201.         @Override
  202.         public TimerService timerService() {
  203.             return timerService;
  204.         }
  205.  
  206.         @Override
  207.         public long currentProcessingTime() {
  208.             return timerService.currentProcessingTime();
  209.         }
  210.  
  211.         @Override
  212.         public long currentWatermark() {
  213.             return timerService.currentWatermark();
  214.         }
  215.  
  216.         @Override
  217.         public <X> void output(OutputTag<X> outputTag, X value) {
  218.             checkArgument(outputTag != null, "OutputTag must not be null.");
  219.             output.collect(outputTag, new StreamRecord<>(value, element.getTimestamp()));
  220.         }
  221.  
  222.         @Override
  223.         public <K, V> ReadOnlyBroadcastState<K, V> getBroadcastState(
  224.                 MapStateDescriptor<K, V> stateDescriptor) {
  225.             Preconditions.checkNotNull(stateDescriptor);
  226.  
  227.             stateDescriptor.initializeSerializerUnlessSet(config);
  228.             ReadOnlyBroadcastState<K, V> state =
  229.                     (ReadOnlyBroadcastState<K, V>) states.get(stateDescriptor);
  230.             if (state == null) {
  231.                 throw new IllegalArgumentException(
  232.                         "The requested state does not exist. "
  233.                                 + "Check for typos in your state descriptor, or specify the state descriptor "
  234.                                 + "in the datastream.broadcast(...) call if you forgot to register it.");
  235.             }
  236.             return state;
  237.         }
  238.  
  239.         @Override
  240.         @SuppressWarnings("unchecked")
  241.         public KS getCurrentKey() {
  242.             return (KS) CoBroadcastWithKeyedOperator.this.getCurrentKey();
  243.         }
  244.     }
  245.  
  246.     private class OnTimerContextImpl
  247.             extends KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT>.OnTimerContext {
  248.  
  249.         private final ExecutionConfig config;
  250.  
  251.         private final Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> states;
  252.  
  253.         private final TimerService timerService;
  254.  
  255.         private TimeDomain timeDomain;
  256.  
  257.         private InternalTimer<KS, VoidNamespace> timer;
  258.  
  259.         OnTimerContextImpl(
  260.                 final ExecutionConfig executionConfig,
  261.                 final KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> function,
  262.                 final Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> broadcastStates,
  263.                 final TimerService timerService) {
  264.  
  265.             function.super();
  266.             this.config = Preconditions.checkNotNull(executionConfig);
  267.             this.states = Preconditions.checkNotNull(broadcastStates);
  268.             this.timerService = Preconditions.checkNotNull(timerService);
  269.         }
  270.  
  271.         @Override
  272.         public Long timestamp() {
  273.             checkState(timer != null);
  274.             return timer.getTimestamp();
  275.         }
  276.  
  277.         @Override
  278.         public TimeDomain timeDomain() {
  279.             checkState(timeDomain != null);
  280.             return timeDomain;
  281.         }
  282.  
  283.         @Override
  284.         public KS getCurrentKey() {
  285.             return timer.getKey();
  286.         }
  287.  
  288.         @Override
  289.         public TimerService timerService() {
  290.             return timerService;
  291.         }
  292.  
  293.         @Override
  294.         public long currentProcessingTime() {
  295.             return timerService.currentProcessingTime();
  296.         }
  297.  
  298.         @Override
  299.         public long currentWatermark() {
  300.             return timerService.currentWatermark();
  301.         }
  302.  
  303.         @Override
  304.         public <X> void output(OutputTag<X> outputTag, X value) {
  305.             checkArgument(outputTag != null, "OutputTag must not be null.");
  306.             output.collect(outputTag, new StreamRecord<>(value, timer.getTimestamp()));
  307.         }
  308.  
  309.         @Override
  310.         public <K, V> ReadOnlyBroadcastState<K, V> getBroadcastState(
  311.                 MapStateDescriptor<K, V> stateDescriptor) {
  312.             Preconditions.checkNotNull(stateDescriptor);
  313.  
  314.             stateDescriptor.initializeSerializerUnlessSet(config);
  315.             ReadOnlyBroadcastState<K, V> state =
  316.                     (ReadOnlyBroadcastState<K, V>) states.get(stateDescriptor);
  317.             if (state == null) {
  318.                 throw new IllegalArgumentException(
  319.                         "The requested state does not exist. "
  320.                                 + "Check for typos in your state descriptor, or specify the state descriptor "
  321.                                 + "in the datastream.broadcast(...) call if you forgot to register it.");
  322.             }
  323.             return state;
  324.         }
  325.     }
  326. }

在分析 CoBroadcastWithKeyedOperator 源码之前，先来看一下 CoBroadcastWithKeyedOperator 的 UML 图。

CoBroadcastWithKeyedOperator UML 图

185dd36f001f0855ef8bd41539ffaa4c.png
CoBroadcastWithKeyedOperator

可以看到 CoBroadcastWithKeyedOperator 实现了 TwoInputStreamOperator 这个接口，从命名上就能知道，这是一个具有两个输入流的 StreamOperator 接口，因为 CoBroadcastWithKeyedOperator 的上游连接的是两个数据流，所以就实现了这个接口，下面再来看一下 TwoInputStreamOperator 的源码。

TwoInputStreamOperator 源码

  1. /**
  2.  * Interface for stream operators with two inputs. Use {@link
  3.  * org.apache.flink.streaming.api.operators.AbstractStreamOperator} as a base class if you want to
  4.  * implement a custom operator.
  5.  *
  6.  * @param <IN1> The input type of the operator
  7.  * @param <IN2> The input type of the operator
  8.  * @param <OUT> The output type of the operator
  9.  */
  10. @PublicEvolving
  11. public interface TwoInputStreamOperator<IN1, IN2, OUT> extends StreamOperator<OUT> {
  12.  
  13.     /**
  14.      * Processes one element that arrived on the first input of this two-input operator. This method
  15.      * is guaranteed to not be called concurrently with other methods of the operator.
  16.      */
  17.     void processElement1(StreamRecord<IN1> element) throws Exception;
  18.  
  19.     /**
  20.      * Processes one element that arrived on the second input of this two-input operator. This
  21.      * method is guaranteed to not be called concurrently with other methods of the operator.
  22.      */
  23.     void processElement2(StreamRecord<IN2> element) throws Exception;
  24.  
  25. }

TwoInputStreamOperator 接口里面定义了两个方法，其中 processElement1 是用来处理非广播流的数据，processElement2 是用来处理广播流的数据。

接着回到 CoBroadcastWithKeyedOperator 的 open 方法，首先会初始化 broadcastStates，用来保存 MapStateDescriptor -> BroadcastState 的映射关系，然后初始化 ReadWriteContextImpl 和 ReadOnlyContextImpl 对象，顾名思义 ReadWriteContextImpl 是既可以读也可以写状态，ReadOnlyContextImpl  是只能读状态，不能写状态，在 open 方法里面还有一个重要的事情，就是初始化广播状态的实现类。

getBroadcastState 源码

  1. public <K, V> BroadcastState<K, V> getBroadcastState(
  2.         final MapStateDescriptor<K, V> stateDescriptor) throws StateMigrationException {
  3.  
  4.     Preconditions.checkNotNull(stateDescriptor);
  5.     String name = Preconditions.checkNotNull(stateDescriptor.getName());
  6.  
  7.     BackendWritableBroadcastState<K, V> previous =
  8.             (BackendWritableBroadcastState<K, V>) accessedBroadcastStatesByName.get(name);
  9.  
  10.     if (previous != null) {
  11.         checkStateNameAndMode(
  12.                 previous.getStateMetaInfo().getName(),
  13.                 name,
  14.                 previous.getStateMetaInfo().getAssignmentMode(),
  15.                 OperatorStateHandle.Mode.BROADCAST);
  16.         return previous;
  17.     }
  18.  
  19.     stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig());
  20.     TypeSerializer<K> broadcastStateKeySerializer =
  21.             Preconditions.checkNotNull(stateDescriptor.getKeySerializer());
  22.     TypeSerializer<V> broadcastStateValueSerializer =
  23.             Preconditions.checkNotNull(stateDescriptor.getValueSerializer());
  24.  
  25.     BackendWritableBroadcastState<K, V> broadcastState =
  26.             (BackendWritableBroadcastState<K, V>) registeredBroadcastStates.get(name);
  27.  
  28.     if (broadcastState == null) {
  29.         broadcastState =
  30.                 new HeapBroadcastState<>(
  31.                         new RegisteredBroadcastStateBackendMetaInfo<>(
  32.                                 name,
  33.                                 OperatorStateHandle.Mode.BROADCAST,
  34.                                 broadcastStateKeySerializer,
  35.                                 broadcastStateValueSerializer));
  36.         registeredBroadcastStates.put(name, broadcastState);
  37.     } else {
  38.         // has restored state; check compatibility of new state access
  39.  
  40.         checkStateNameAndMode(
  41.                 broadcastState.getStateMetaInfo().getName(),
  42.                 name,
  43.                 broadcastState.getStateMetaInfo().getAssignmentMode(),
  44.                 OperatorStateHandle.Mode.BROADCAST);
  45.  
  46.         RegisteredBroadcastStateBackendMetaInfo<K, V> restoredBroadcastStateMetaInfo =
  47.                 broadcastState.getStateMetaInfo();
  48.  
  49.         // check whether new serializers are incompatible
  50.         TypeSerializerSchemaCompatibility<K> keyCompatibility =
  51.                 restoredBroadcastStateMetaInfo.updateKeySerializer(broadcastStateKeySerializer);
  52.         if (keyCompatibility.isIncompatible()) {
  53.             throw new StateMigrationException(
  54.                     "The new key typeSerializer for broadcast state must not be incompatible.");
  55.         }
  56.  
  57.         TypeSerializerSchemaCompatibility<V> valueCompatibility =
  58.                 restoredBroadcastStateMetaInfo.updateValueSerializer(
  59.                         broadcastStateValueSerializer);
  60.         if (valueCompatibility.isIncompatible()) {
  61.             throw new StateMigrationException(
  62.                     "The new value typeSerializer for broadcast state must not be incompatible.");
  63.         }
  64.  
  65.         broadcastState.setStateMetaInfo(restoredBroadcastStateMetaInfo);
  66.     }
  67.  
  68.     accessedBroadcastStatesByName.put(name, broadcastState);
  69.     return broadcastState;
  70. }

getBroadcastState 方法主要就是初始化 HeapBroadcastState 对象，也就是广播状态的具体实现类，再来看一下 HeapBroadcastState 源码。

HeapBroadcastState 源码

  1. /**
  2.  * A {@link BroadcastState Broadcast State} backed a heap-based {@link Map}.
  3.  *
  4.  * @param <K> The key type of the elements in the {@link BroadcastState Broadcast State}.
  5.  * @param <V> The value type of the elements in the {@link BroadcastState Broadcast State}.
  6.  */
  7. public class HeapBroadcastState<K, V> implements BackendWritableBroadcastState<K, V> {
  8.  
  9.     /** Meta information of the state, including state name, assignment mode, and serializer. */
  10.     private RegisteredBroadcastStateBackendMetaInfo<K, V> stateMetaInfo;
  11.  
  12.     /** The internal map the holds the elements of the state. */
  13.     private final Map<K, V> backingMap;
  14.  
  15.     /** A serializer that allows to perform deep copies of internal map state. */
  16.     private final MapSerializer<K, V> internalMapCopySerializer;
  17.  
  18.     HeapBroadcastState(RegisteredBroadcastStateBackendMetaInfo<K, V> stateMetaInfo) {
  19.         this(stateMetaInfo, new HashMap<>());
  20.     }
  21.  
  22.     private HeapBroadcastState(
  23.             final RegisteredBroadcastStateBackendMetaInfo<K, V> stateMetaInfo,
  24.             final Map<K, V> internalMap) {
  25.  
  26.         this.stateMetaInfo = Preconditions.checkNotNull(stateMetaInfo);
  27.         this.backingMap = Preconditions.checkNotNull(internalMap);
  28.         this.internalMapCopySerializer =
  29.                 new MapSerializer<>(
  30.                         stateMetaInfo.getKeySerializer(), stateMetaInfo.getValueSerializer());
  31.     }
  32.  
  33.     private HeapBroadcastState(HeapBroadcastState<K, V> toCopy) {
  34.         this(
  35.                 toCopy.stateMetaInfo.deepCopy(),
  36.                 toCopy.internalMapCopySerializer.copy(toCopy.backingMap));
  37.     }
  38.  
  39.     @Override
  40.     public void setStateMetaInfo(RegisteredBroadcastStateBackendMetaInfo<K, V> stateMetaInfo) {
  41.         this.stateMetaInfo = stateMetaInfo;
  42.     }
  43.  
  44.     @Override
  45.     public RegisteredBroadcastStateBackendMetaInfo<K, V> getStateMetaInfo() {
  46.         return stateMetaInfo;
  47.     }
  48.  
  49.     @Override
  50.     public HeapBroadcastState<K, V> deepCopy() {
  51.         return new HeapBroadcastState<>(this);
  52.     }
  53.  
  54.     @Override
  55.     public void clear() {
  56.         backingMap.clear();
  57.     }
  58.  
  59.     @Override
  60.     public String toString() {
  61.         return "HeapBroadcastState{"
  62.                 + "stateMetaInfo="
  63.                 + stateMetaInfo
  64.                 + ", backingMap="
  65.                 + backingMap
  66.                 + ", internalMapCopySerializer="
  67.                 + internalMapCopySerializer
  68.                 + '}';
  69.     }
  70.  
  71.     @Override
  72.     public long write(FSDataOutputStream out) throws IOException {
  73.         long partitionOffset = out.getPos();
  74.  
  75.         DataOutputView dov = new DataOutputViewStreamWrapper(out);
  76.         dov.writeInt(backingMap.size());
  77.         for (Map.Entry<K, V> entry : backingMap.entrySet()) {
  78.             getStateMetaInfo().getKeySerializer().serialize(entry.getKey(), dov);
  79.             getStateMetaInfo().getValueSerializer().serialize(entry.getValue(), dov);
  80.         }
  81.  
  82.         return partitionOffset;
  83.     }
  84.  
  85.     @Override
  86.     public V get(K key) {
  87.         return backingMap.get(key);
  88.     }
  89.  
  90.     @Override
  91.     public void put(K key, V value) {
  92.         backingMap.put(key, value);
  93.     }
  94.  
  95.     @Override
  96.     public void putAll(Map<K, V> map) {
  97.         backingMap.putAll(map);
  98.     }
  99.  
  100.     @Override
  101.     public void remove(K key) {
  102.         backingMap.remove(key);
  103.     }
  104.  
  105.     @Override
  106.     public boolean contains(K key) {
  107.         return backingMap.containsKey(key);
  108.     }
  109.  
  110.     @Override
  111.     public Iterator<Map.Entry<K, V>> iterator() {
  112.         return backingMap.entrySet().iterator();
  113.     }
  114.  
  115.     @Override
  116.     public Iterable<Map.Entry<K, V>> entries() {
  117.         return backingMap.entrySet();
  118.     }
  119.  
  120.     @Override
  121.     public Iterable<Map.Entry<K, V>> immutableEntries() {
  122.         return Collections.unmodifiableSet(backingMap.entrySet());
  123.     }
  124. }

HeapBroadcastState 的代码比较简单，主要是对状态的读写操作，本质就是在操作 HashMap。

接着回到 CoBroadcastWithKeyedOperator 的 processElement1 方法里用的是 ReadOnlyContextImpl，processElement2 方法里用的是 ReadWriteContextImpl，换句话说，只有在广播侧才可以修改状态，在非广播侧不能修改状态，这里对应了上面的第二个问题。

虽然在广播侧和非广侧都可以获取到状态，但是 getBroadcastState 方法的返回值是不一样的。

BroadcastState & ReadOnlyBroadcastState UML 图

3ac66f5955e5f6302964e784a1d829ea.png

HeapBroadcastState

BroadcastState 接口继承了 ReadOnlyBroadcastState 接口又继承了 State 接口，BroadcastState 接口的唯一实现类是 HeapBroadcastState，从名字上就能看出广播状态是存储在 JVM 堆内存上的。底层就是一个 Map，上图中的 backingMap 就是用来保存状态数据的，这里对应了上面的第三个问题。

为了进一步解释上面的第二个问题，下面补充一个具体的场景来说明。

举例说明

f781512826cbf00008973662466f5de2.png

BroadcastStream

我们来看上图中的场景，A 流读取 Kafka 的数据然后经过 keyby 返回一个 KeyedStream，B 流读取 mysql 的数据用于广播流返回一个 BroadcastStream，B 流有两条数据分别是 flink,spark，然后会广播到下游的每一个 subtask 上去，此时下游的 subtask-0,subtask-1 就拥有了广播状态中的 flink,spark 两条数据，这个时候往 Kafka 里写入两条数据分别是 flink 和 hive,经过 keyby 操作，flink 被分配到了下游的 subtask-0 上，hive 被分配到了 subtask-1 上，很明显 flink 这条数据可以和广播流数据关联上，hive 这条数据则关联不上，此时，如果在非广播侧也就是 A 流侧修改了状态，比如把 flink, hive 添加到了状态里面，此时 subtask-0 和 subtask-1 上的广播状态数据就会出现不一致的情况，所以，为了保证 operator 的所有并发实例持有的广播状态的一致性，在设计的时候就禁止在非广播侧修改状态。

总结

Broadcast State 是 Operator State 的一种特殊类型。主要是用来解决低吞吐量的流（小数据量）和另一个原始数据流关联的场景，广播状态必须定义为 Map 结构，并且只能在广播流侧修改状态，非广播流侧只能获取状态，不能修改状态。广播状态只能保存在堆内存中，所以在使用广播状态的时候需要给 TM 设置足够的内存，本文主要从源码的角度解释了 Flink 这么设计的原因，让大家对广播流状态有了更加深入的理解。

推荐阅读

!! Flink 任务实时监控最佳实践

Flink on yarn 实时日志收集最佳实践

Flink 1.14.0 全新的 Kafka Connector

Flink 1.14.0 消费 kafka 数据自定义反序列化类

Flink SQL JSON Format 源码解析

Flink 通过 State Processor API 实现状态的读取和写入

Flink 源码分析之 Client 端启动流程分析

Flink Print SQL Connector 添加随机取样功能

Flink on yarn 远程调试源码

IDEA 中使用 Big Data Tools 连接大数据组件

8a22bb31c37cf05f3ca62a0490155df2.png

图片

如果你觉得文章对你有帮助,麻烦点一下 赞 和 在看 吧,你的支持是我创作的最大动力.

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

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

  

闽ICP备14008679号