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();
}
DAG
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cxsxw4h6-1643447157377)(broadcast.assets/image-20220129165306583.png)]
数据流向：
蓝色：rebalance
红色：broadcast
绿色：forward
橙色：hash
https://www.cnblogs.com/jmx-bigdata/p/13708873.html
