如何在键控广播过程中控制 flink 发送输出到侧输出函数

Question

我正在尝试根据一组规则验证数据流以检测 flink 中的模式，方法是使用一组规则验证数据流，我使用 for 循环收集 map 中的所有模式并在 processElement fn 中迭代它以查找模式示例代码如下

映射状态描述符和侧输出流如下所示

public static final MapStateDescriptor<String, String> ruleSetDescriptor =
new MapStateDescriptor<String, String>("RuleSet", BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO);
public final static OutputTag<Tuple2<String, String>> unMatchedSideOutput =
new OutputTag<Tuple2<String, String>>(
"unmatched-side-output") {
};

过程功能和广播功能如下：

@Override
public void processElement(Tuple2<String, String> inputValue, ReadOnlyContext ctx,
Collector<Tuple2<String,
String>> out) throws Exception {

for (Map.Entry<String, String> ruleSet:
ctx.getBroadcastState(broadcast.patternRuleDescriptor).immutableEntries()) {
String ruleName = ruleSet.getKey();

//If the rule in ruleset is matched then send output to main stream and break the program
if (this.rule) {
out.collect(new Tuple2<>(inputValue.f0, inputValue.f1));
break;
}
}
// Writing output to sideout if no rule is matched 
ctx.output(Output.unMatchedSideOutput, new Tuple2<>("No Rule Detected", inputValue.f1));
}

@Override
public void processBroadcastElement(Tuple2<String, String> ruleSetConditions, Context ctx, Collector<Tuple2<String,String>> out) throws Exception {
ctx.getBroadcastState(broadcast.ruleSetDescriptor).put(ruleSetConditions.f0,
ruleSetConditions.f1);
}

我能够检测到模式，但我也得到了侧输出，因为我正在尝试一个接一个地迭代规则，如果我最后存在匹配的规则，程序正在将输出发送到旁输出，因为初始规则集不匹配。如果不满足任何规则，我只想打印一次侧输出，我是 flink 的新手，请帮助我如何实现它。

Answer 1

在我看来，你想做更多类似的事情：

@Override
public void processElement(Tuple2<String, String> inputValue, ReadOnlyContext ctx, Collector<Tuple2<String, String>> out) throws Exception {
transient boolean matched = false;
for (Map.Entry<String, String> ruleSet:
ctx.getBroadcastState(broadcast.patternRuleDescriptor).immutableEntries()) {
String ruleName = ruleSet.getKey();
if (this.rule) {
matched = true;
out.collect(new Tuple2<>(inputValue.f0, inputValue.f1));
break;
}
}
// Writing output to sideout if no rule was matched
if (!matched) {
ctx.output(Output.unMatchedSideOutput, new Tuple2<>("No Rule Detected", inputValue.f1));
}
}