我正在尝试使用一组规则验证JSONObject,如果json与一组规则匹配,是它将返回匹配的规则,而如果JSON与否,它将返回一个JSONObject到Sideoutput,所有这些都在ProcessFuntion中处理,我正在获取主要输出但无法捕获侧面输出
SideOutput流的定义如下
public final static OutputTag<org.json.JSONObject> unMatchedJSONSideOutput = new OutputTag<org.json.JSONObject>(
"unmatched-side-output") {};
ProcessFunction定义如下
public class RuleFilter extends ProcessFunction<Tuple2<String,org.json.JSONObject>,Tuple2<String,org.json.JSONObject>> {
@Override
public void processElement(Tuple2<String, org.json.JSONObject> value,
ProcessFunction<Tuple2<String, org.json.JSONObject>, Tuple2<String, org.json.JSONObject>>.Context ctx,
Collector<Tuple2<String, org.json.JSONObject>> out) throws Exception {
if(this.value.matches((value.f1))) {
out.collect(new Tuple2<String, org.json.JSONObject>(value.f0,value.f1));
}else {
ctx.output(RuleMatching.unMatchedJSONSideOutput,value.f1);
}
}
}
我正在打印主要的数据流输出,如下所示
DataStream<Tuple2<String, org.json.JSONObject>> matchedJSON =
inputSignal.map(new MapFunction<org.json.JSONObject, Tuple2<String, org.json.JSONObject>>() {
@Override
public Tuple2<String, org.json.JSONObject> map(org.json.JSONObject input) throws Exception {
return new Tuple2<>(value, input);
}
}).process(new RuleFilter()).print("MatchedJSON=>");
matchedJSON .print("matchedJSON=>");
我正在打印如下的Sideoutput
DataStream<org.json.JSONObject> unmatchedJSON =
((SingleOutputStreamOperator<org.json.JSONObject>) matchedJSON.map(new MapFunction<Tuple2<String, org.json.JSONObject>, org.json.JSONObject>() {
@Override
public org.json.JSONObject map(Tuple2<String, org.json.JSONObject> value) throws Exception {
return value.f1;
}
})).getSideOutput(unMatchedJSONSideOutput );
unmatchedJSON.print("unmatchedJSON=>");
主流正在打印输出,但是sideoutput没有为无效的json打印,请帮助解决问题
DataStream<org.json.JSONObject> unmatchedJSON =
((SingleOutputStreamOperator<org.json.JSONObject>) matchedJSON.map(...))
.getSideOutput(unMatchedJSONSideOutput);
您应该直接在getSideOutput
上调用matchedJSON
,而不是在对其施加MapFunction
的结果上。仅ProcessFunction
可以有侧面输出,它需要直接来自ProcessFunction
。您欺骗了编译器通过从映射强制转换输出流来接受此操作,但是运行时对此无能为力。