Я пытаюсь проверить JSONObject с набором правил, если json соответствует набору правил, он вернет соответствующее правило, а JSONObject, если нет, он вернет JSONObject в Sideoutput, все это обрабатывается в ProcessFuntion, я получаю основной вывод, но не удалось захватить побочный вывод
SideOutput Stream определяется следующим образом
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, как показано ниже
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=>");
Основной поток выводит на печать вывод, но побочный вывод не выводится из-за недопустимого json, пожалуйста, помогите в решении проблемы.