16、Flink 的广播状态 (Broadcast State) 代码示例_flink 广播 例子

1、Broadcast State 案例

规则流：1,a,b [规则名1 规则为 a 或 b]
图形流：green,a [绿色 a]

问题：如果规则流先于数据流则匹配不上=>此时缓冲数据流中的数据【如果规则流为null】

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.*;

public class _06_BroadcastState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 接收广播的规则数据
        SingleOutputStreamOperator<_06_Rule> ruleStream = env.socketTextStream("localhost", 8888)
                .map(new MapFunction<String, _06_Rule>() {
                    @Override
                    public _06_Rule map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new _06_Rule(fields[0], new Tuple2<>(fields[1], fields[2]));
                    }
                });

        // 一个 map descriptor，它描述了用于存储规则名称与规则本身的 map 存储结构
        MapStateDescriptor<String, _06_Rule> ruleStateDescriptor = new MapStateDescriptor<>(
                "RulesBroadcastState",
                BasicTypeInfo.STRING_TYPE_INFO,
                TypeInformation.of(new TypeHint<_06_Rule>() {
                }));

        // 广播流，广播规则并且创建 broadcast state
        BroadcastStream<_06_Rule> ruleBroadcastStream = ruleStream
                .broadcast(ruleStateDescriptor);


        // 接收图形数据
        SingleOutputStreamOperator<_06_Shape> shapeStream = env.socketTextStream("localhost", 9999)
                .map(new MapFunction<String, _06_Shape>() {
                    @Override
                    public _06_Shape map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new _06_Shape(fields[0], fields[1]);
                    }
                });

        shapeStream.keyBy(_06_Shape::getColour)
                .connect(ruleBroadcastStream)
                .process(new KeyedBroadcastProcessFunction<String, _06_Shape, _06_Rule, String>() {
//                    private transient ValueState<List<_06_Shape>> dataBuffer;
                    private transient ListState<_06_Shape> dataBuffer;

                    @Override
                    public void open(Configuration parameters) throws Exception {
//                        ValueStateDescriptor<List<_06_Shape>> dataListStateDescriptor = new ValueStateDescriptor<>("dataBuffer", TypeInformation.of(new TypeHint<List<_06_Shape>>() {
//                        }));
//
//                        dataBuffer = getRuntimeContext().getState(dataListStateDescriptor);

                        ListStateDescriptor<_06_Shape> listStateDescriptor = new ListStateDescriptor<>("dataBuffer", _06_Shape.class);
                        dataBuffer = getRuntimeContext().getListState(listStateDescriptor);
                    }

                    @Override
                    public void processElement(_06_Shape value, KeyedBroadcastProcessFunction<String, _06_Shape, _06_Rule, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        // 获取广播的规则数据
                        System.out.println("输入的数据颜色为=>" + value.getColour() + "，类型为=>" + value.getType());
                        ReadOnlyBroadcastState<String, _06_Rule> broadcastState = ctx.getBroadcastState(ruleStateDescriptor);

                        Iterator<Map.Entry<String, _06_Rule>> iterator = broadcastState.immutableEntries().iterator();

                        if (iterator.hasNext()) {
                            // 使用 ValueState
                            // 先从缓存中读取数据进行匹配
//                            List<_06_Shape> shapeList = dataBuffer.value();
                            // 多并行度时，防止某个并行度无数据导致报错
//                            if (shapeList != null) {
//                                if (!shapeList.isEmpty()) {
//                                    for (_06_Shape shape : shapeList) {
//                                        System.out.println("被缓冲的数据开始进行处理=>" + shape);
//                                        // 从事件数据中继续匹配
//                                        while (iterator.hasNext()) {
//                                            Map.Entry<String, _06_Rule> rule = iterator.next();
//                                            if (Objects.equals(rule.getValue().getRule().f0, shape.getType()) || Objects.equals(rule.getValue().getRule().f1, shape.getType())) {
//                                                out.collect("匹配上的数据为=>" + value + "匹配上的规则名称为=>" + rule.getValue().getRuleName());
//                                            }
//                                        }
//                                    }
//
//                                    shapeList.clear();
//                                }
//                            }

                            // 使用 ListState
                            Iterator<_06_Shape> dataIterator = dataBuffer.get().iterator();
                            while (dataIterator.hasNext()){
                                _06_Shape shape = dataIterator.next();
                                System.out.println("被缓冲的数据开始进行处理=>" + shape);

                                while (iterator.hasNext()) {
                                    Map.Entry<String, _06_Rule> rule = iterator.next();
                                    if (Objects.equals(rule.getValue().getRule().f0, value.getType()) || Objects.equals(rule.getValue().getRule().f1, value.getType())) {
                                        out.collect("匹配上的数据为=>" + value + "匹配上的规则名称为=>" + rule.getValue().getRuleName());
                                    }
                                }

                                dataIterator.remove();
                            }

                            // 从事件数据中继续匹配
                            while (iterator.hasNext()) {
                                Map.Entry<String, _06_Rule> rule = iterator.next();
                                if (Objects.equals(rule.getValue().getRule().f0, value.getType()) || Objects.equals(rule.getValue().getRule().f1, value.getType())) {
                                    out.collect("匹配上的数据为=>" + value + "匹配上的规则名称为=>" + rule.getValue().getRuleName());
                                }
                            }
                        } else {
                            System.out.println("此时规则流中无规则，先缓冲数据流");
                            // 使用 listState
                            dataBuffer.add(value);

                            // 使用 valueState
//                            List<_06_Shape> shapeList = dataBuffer.value();
//                            if (shapeList == null) {
//                                shapeList = new ArrayList<>();
//                            }
//                            shapeList.add(value);
//                            dataBuffer.update(shapeList);
                        }
                    }

                    @Override
                    public void processBroadcastElement(_06_Rule value, KeyedBroadcastProcessFunction<String, _06_Shape, _06_Rule, String>.Context ctx, Collector<String> out) throws Exception {
                        // 获取广播流输入的数据，存入广播状态
                        System.out.println("输入的规则名称为=>" + value.getRuleName() + "，规则为=>" + value.getRule());
                        BroadcastState<String, _06_Rule> broadcastState = ctx.getBroadcastState(ruleStateDescriptor);
                        broadcastState.put(value.getRuleName(), value);
                    }
                })
                .print();

        env.execute();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

2、Pojo 类

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;

import java.io.Serializable;

@NoArgsConstructor
@AllArgsConstructor
@Data
public class _06_Rule implements Serializable {
    private String ruleName;
    private Tuple2<String,String> rule;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.io.Serializable;

@AllArgsConstructor
@NoArgsConstructor
@Data
public class _06_Shape implements Serializable {
    private String colour;
    private String type;
}
1
2
3
4
5
6
7
8
9
10
11
12
13