Apache Flink:如何处理三个流

Question

我想在一个操作符中接收和处理三个流。例如，Storm中实现的代码如下：

builder.setBolt("C_bolt", C_bolt(), parallelism_hint) .fieldsGrouping("A_bolt", "TRAINING", new Fields("word")) .fieldsGrouping("B_bolt", "ANALYSIS", new Fields("word")) .allGrouping("A_bolt", "SUM");

在Flink中，实现了对SUM stream(A_bolt's SideOutput)和TRAINING stream(A_bolt)的处理：

SingleOutputStreamOperator<Tuple3<String, Integer, Boolean>> A_bolt;
DataStream<Tuple2<Integer, Integer>> Sum = A_bolt.getSideOutput(outputTag).broadcast();
DataStream<Tuple3<String, String, Integer>> B_bolt;
DataStream<String> C_bolt= A_bolt
.keyBy(new KeySelector<Tuple3<String,Integer,Boolean>, String>() {
@Override
public String getKey(Tuple3<String,Integer,Boolean> in) throws Exception {
return in.f0;
}
})
.connect(Sum)
.flatMap(new Process())
.setParallelism(parallelism);

但是我不知道如何添加ANALYSIS stream(B_bolt)。谢谢你的帮助。

Answer 1

Flink只支持一个输入和两个输入流运算符。您的选择是：

1. 使用union((创建一个合并流，其中包含所有三个流中的所有元素(它们必须都是相同的类型，但您可以使用"两者之一"来帮助实现这一点(
2. 使用coFlatMap组合两个流后，将初步结果连接到第三个流，使用另一个coFlatMap(或coProcessFunction(完成处理

或者在您的情况下，这两种技术的组合可能更可取。