上级： https://www.cnblogs.com/hackerxiaoyon/p/12747387.html

状态启动描述了操作算子的状态，在恢复时，该状态是均匀分布在操作算子的并行任务之间，或者是合并的，整个状态用于初始化恢复的并行任务。

 

支持状态算子的第三类是广播状态。广播状态就是在我们接入一个流的时候需要广播变量到下游所有任务，这些状态存储在本地被用于处理所有来的元素从其他流。例子，广播变量可以自然的适应，假设一个低吞吐流包含一个规则集合我们想要评估所有来自其他流的数据应用此规则。基于以上案例，广播状态不同于其他状态算子的区别在于：

它有一个映射格式

针对特定算子有效作为广播流和没有广播

这样的算子可以有多个广播状态并且是不同的名字

 

Provided APIs

先举例子：我们将会使用一个事例，一个流存在不同的颜色和形状对象，然后我们想要找到相同颜色的对象根据某个原则。如：一个矩形后跟着一个三角形。我们假设这些有趣的模型会随着时间而发生演变。

 

第一条流包含颜色和形状属性，其他流包含规则。

我们拿到流进行keyby根据color，因为我们想要拿到相同颜色在一起。

我们设置规则，规则流被广播都下游都所有任务，这些任务存储在本地将规则，然后从items流来进行评估。接下来快照会广播规则流，使用MapStateDescriptor，创建广播状态存储规则。

 

最后，为了评估规则流从item流中，我们需要：

1、连接两个流

2、具体我们的匹配逻辑

 

连接一个流用BroadcastStream可以通过connect()在非广播流上，需要BroadcastStream作为参数。会返回一个BroadcastConnectedStream，然后我们可以调用process()指定一个具体CoProcessFunction类型。这个函数包含我们的匹配逻辑。函数的确切类型依赖于非广播流的类型：

1、如果是keyed的类型，函数就是KeyedBroadcastProcessFunction

2、如果是non-keyed类型，函数就是BroadcastProcessFunction

 

BroadcastProcessFunction and KeyedBroadcastProcessFunction

 

此例子处理方法有两个处理方法需要实现；processBroadElement()负责处理广播流数据，processElement()用于处理非广播。

第一件事需要注意的两个函数需要实现processBroadcastElement()去处理广播中的数据，processElement()处理非广播的数据。

两个方法在内容提供上不同。非广播有一个ReadOnlyContext，广播的一边有Context。

 

上下文：

1、获取广播状态：ctx.getBroadcastState(MapStateDescriptor<K,V> stateDescriptor)

2、允许查询元素的时间戳：ctx.timestamp()

3、获取当前水印；ctx.currentWatermark()

4、获取当前处理时间：ctx.currentProcessingTime()

5、发送数据到单边流:ctx.output(OutputTag<X> outputTag,X value)

 

stateDescriptor 在getBroadcastState() 中和代码的.broadcast(ruleStateDescriptor)一样。

广播流和非广播流重的广播状态不同，一个是read_write，另一个是read_only。原因是flink没有跨任务交流。因此，为了保证内容在我们广播状态一致在我们并行算子，我们广播单边用read_write，这样可以在所有任务看到相同元素，我们需要计算每一个来的元素使得跨所有任务达到一致。忽略这个规则将会破坏状态的一致性保障，导致不一致性经常与debug结果不一样。

 

注意：⚠️processBroadcast()逻辑实现必须有确定相同的行为在所有并行实例！

最后，事实上KeyedBroadcastProcessFunction操作在key流上，暴露了一些BroadcastProcessFunction无效的函数。

1、processElement()里面的ReadOnlyContext可以提供定时服务，允许注册事件/处理事件定时器。当定时器触发，onTimer()会在OnTimerContext被触发。

2、processBroadcastElement()里面的Context包含applyToKeyedState(StateDescriptor<S,VS> stateDescriptor,KeyedStateFunction<KS,S> function).允许注册一个KeyedStateFunction应到所有关键字的状态。

 

 

Important Considerations 

使用广播状态需要注意的一些重要事情：

1、不存在跨任务通信：作为最早的状态，这也是为什么只有广播单边可以修改广播状态那些的原因。此外，用户必须确定对于所有任务的每条流元素修改状态内容是同一方法。否则，不同任务有不同内容，导致不一致的结果。

2、广播状态事件顺序在不同状态中可能不同：虽然广播流的元素保证可以达到所有的下流任务，元素可能达到每个任务的顺序不同。因此元素状态更新一定不能依赖事件顺序。

3、所有任务检查点他们的广播状态：虽然所有任务有相同的元素在广播状态当一个检查点替换时候，所有任务检查他们的广播状态，而不是仅仅一个。这是一个读取策略避免所有任务从一个文件读取在恢复期间，避免热点。尽管这样做增加了检查点的因素。flink保障数据恢复重启作业没有重复和没有丢失。相同活着更小的并行度恢复，每个作业读取它的检查点状态，其他任务读取上一个任务检查点轮询方式。。

4、无rocksDB状态后台：广播状态在运行时保持在内存中，内存配置应相应进行。这适用于所有的算子状态。

 

代码

 

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

DataStreamSource<String> localhost = env.socketTextStream("localhost", 8888, "\n");
SingleOutputStreamOperator<Item> source = localhost.map(new MapFunction<String, Item>() {
    @Override
    public Item map(String s) throws Exception {
        String[] split = s.split(",");
        return new Item(Integer.parseInt(split[0]), split[1]);
    }
});

DataStreamSource<String> ruleStream = env.fromElements("");


KeyedStream<Item, Integer> keyedStream = source.keyBy(new KeySelector<Item, Integer>() {
    @Override
    public Integer getKey(Item item) throws Exception {
        return item.getColor();
    }
});

MapStateDescriptor<String, Rule> ruleMapStateDescriptor = new MapStateDescriptor<>(
        "RulesBroadcastState",
        BasicTypeInfo.STRING_TYPE_INFO,
        TypeInformation.of(new TypeHint<Rule>() {
        }));

BroadcastStream<String> ruleBroadcastStream = ruleStream.broadcast(ruleMapStateDescriptor);

DataStream<String> output = keyedStream.connect(ruleBroadcastStream)
        .process(new KeyedBroadcastProcessFunction<Object, Item, String, String>() {
            @Override
            public void processElement(Item value, ReadOnlyContext ctx, Collector<String> out) throws Exception {

                ReadOnlyBroadcastState<Object, Object> broadcastState = ctx.getBroadcastState(new MapStateDescriptor<Object, Object>(toString(), Object.class, Object.class));
                Long timestamp = ctx.timestamp();
                long l = ctx.currentWatermark();
                long l1 = ctx.currentProcessingTime();
                //ctx.output();

            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {

            }
        });