broadcast

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/fault-tolerance/broadcast_state/

broadcast state

broadcast Stream

将整个broadcastState完整发送到下游的所有task中，使用如下模型可以方便的对广播流的内容进行实时跟新并和dataStream进行数据处理

//1
 MapStateDescriptor<key, value>> state = new MapStateDescriptor("state", keySerilizer, valueSerilizer>() {}));


dataStream
//2，使用broadcstProcessFunction，join的流必须为broadStream
    .join(broadcastStream)
    .process(new BroadcastProcessFunction<<IN1, IN2, OUT>() {
                    @Override
                    public void processElement(String value, ReadOnlyContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {

//3. 使用
                        //获取broadcastState
                        ReadOnlyBroadcastState<String, Tuple2<String, Integer>> broadcastState = ctx.getBroadcastState(state);
                        //获取广播状态并转为不可变entries
                        Iterable<Map.Entry<String, Tuple2<String, Integer>>> entries = broadcastState.immutableEntries();
                       
                    }

                    @Override
                    public void processBroadcastElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
// 4.更新                    
                     //获取广播状态，并将新跟新的值输入到状态中
                        ctx.getBroadcastState(state).put(value.f0, value);

                    }
                })

e.g：将字符流（流表）和字符权重流（维表）连接，给每个字符匹配对应的权重并输出

demo:

  public static void main(String[] args) throws Exception {

        MapStateDescriptor<String, Tuple2<String, Integer>> state = new MapStateDescriptor("state", Types.STRING, TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {
        }));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<Tuple2<String, Integer>> wordWeight = env.fromElements(Tuple2.of("a", 1), Tuple2.of("b", 1), Tuple2.of("c", 1), Tuple2.of("d", 2), Tuple2.of("e", 1), Tuple2.of("f", 2), Tuple2.of("g", 3));
        DataStreamSource<String> strs = env.socketTextStream("10.164.29.148", 10086);
        BroadcastStream<Tuple2<String, Integer>> broadcast = wordWeight.broadcast(state);

        strs
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        Arrays.stream(value.split("\\s+")).forEach(out::collect);
                    }
                })
            //连接的流必须为广播流才能使用BroadcastProcessFunction()对广播的状态进行访问
                .connect(broadcast)
                .process(new BroadcastProcessFunction<String, Tuple2<String,Integer>, Tuple2<String, Integer>>() {
                    @Override
                    public void processElement(String value, ReadOnlyContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {

                        ReadOnlyBroadcastState<String, Tuple2<String, Integer>> broadcastState = ctx.getBroadcastState(state);
                        Iterable<Map.Entry<String, Tuple2<String, Integer>>> entries = broadcastState.immutableEntries();
//                        entries.forEach(e -> System.out.println(e.getKey()+"_-_" + e.getValue()));
                        /*entries.forEach(e -> {
                            if (e.getKey().equals(value)) {
                                out.collect(Tuple2.of(value, e.getValue().f1));
                            }
                        });*/
                        out.collect(Tuple2.of(value, broadcastState.contains(value) ? broadcastState.get(value).f1 : null));
                    }

                    @Override
                    public void processBroadcastElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
//                        TimeUnit.SECONDS.sleep(2);
                        ctx.getBroadcastState(state).put(value.f0, value);
//                        System.out.println("BC" + value.f0 + "_" + value);
                    }
                })
                .print();

        env.execute();

    }

输入：

a
a
b
c
d
e
f
g
h
a
e
c
d
f
v
a

a

a
b
c
d
f

输出：

5> (a,1)
6> (a,1)
7> (b,1)
8> (c,1)
1> (d,2)
2> (e,1)
3> (f,2)
4> (g,3)
5> (h,null)
6> (a,1)
7> (e,1)
8> (c,1)
1> (d,2)
2> (f,2)
3> (v,null)
4> (a,1)
5> (,null)
6> (a,1)
7> (,null)
8> (a,1)
1> (b,1)
2> (c,1)
3> (d,2)
4> (f,2)

动态改变广播流内容并计算字符值的历史和：

 public static void main(String[] args) throws Exception {
        MapStateDescriptor<String, Tuple2<String, Integer>> state = new MapStateDescriptor("state", Types.STRING, TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {
        }));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
//        DataStreamSource<Tuple2<String, Integer>> wordWeight = env.fromElements(Tuple2.of("a", 1), Tuple2.of("b", 1), Tuple2.of("c", 1), Tuple2.of("d", 2), Tuple2.of("e", 1), Tuple2.of("f", 2), Tuple2.of("g", 3));
        DataStream<Tuple2<String, Integer>> wordWeight = env.socketTextStream("10.164.29.148", 10011)
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] e = value.split("\\s+");
                        if (e.length == 2) {
                            out.collect(Tuple2.of(e[0], Integer.valueOf(e[1])));
                        } else      
                        System.err.println("format exception by: " + value);
                    }
                });
        DataStreamSource<String> strs = env.socketTextStream("10.164.29.148", 10086);
        BroadcastStream<Tuple2<String, Integer>> broadcast = wordWeight.broadcast(state);

        strs
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        Arrays.stream(value.split("\\s+")).forEach(out::collect);
                    }
                })
                .connect(broadcast)
                .process(new BroadcastProcessFunction<String, Tuple2<String,Integer>, Tuple2<String, Integer>>() {
                    @Override
                    public void processElement(String value, ReadOnlyContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {

                        ReadOnlyBroadcastState<String, Tuple2<String, Integer>> broadcastState = ctx.getBroadcastState(state);
                        Iterable<Map.Entry<String, Tuple2<String, Integer>>> entries = broadcastState.immutableEntries();
                        out.collect(Tuple2.of(value, broadcastState.contains(value) ? broadcastState.get(value).f1 : null));
                    }

                    @Override
                    public void processBroadcastElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                        ctx.getBroadcastState(state).put(value.f0, value);
                    }
                })
           		.keyBy(e -> e.f0)
                .reduce((e, ee) -> Tuple2.of(e.f0, e.f1+ee.f1))
                .print();

        env.execute();

    }