尝试在PatternStream上选择时“输入不匹配：需要元组类型”

Question

val pattern : Pattern[TrafficEvent, _] = Pattern.begin[TrafficEvent]("start")
val patternStream = CEP.pattern(stream.javaStream, pattern);

class MyPatternSelectFunction extends PatternSelectFunction[TrafficEvent, TrafficEvent] {
    override def select(pattern : java.util.Map[String, TrafficEvent]) : TrafficEvent ={
        pattern.get("start")
    }
}
val alerts = patternStream.select(new MyPatternSelectFunction())

  Input mismatch: Tuple type expected.
            org.apache.flink.api.java.typeutils.TypeExtractor.validateInputType(TypeExtractor.java:878)
            org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:302)
            org.apache.flink.cep.PatternStream.select(PatternStream.java:64)
            com.demo.DemoTraffic$.main(DemoTraffic.scala:311)

public static DataStream<DemoTraffic.trafficEvent>  getStreamByPattern(DataStream<DemoTraffic.trafficEvent> stream) {
  Pattern<DemoTraffic.trafficEvent, ?> pattern = Pattern.<DemoTraffic.trafficEvent>begin("start");
  PatternStream<DemoTraffic.trafficEvent> patternStream = CEP.pattern(stream, pattern);
  DataStream<DemoTraffic.trafficEvent> rvalue = patternStream.select(new PatternSelectFunction<DemoTraffic.trafficEvent, DemoTraffic.trafficEvent>() {
    @Override
    public DemoTraffic.trafficEvent select(Map<String, DemoTraffic.trafficEvent> pattern) throws Exception {
      return pattern.get("start");
    }
  });
  return rvalue;
}

谢谢

Answer 1

我假设TrafficEvent是一个Scala case类。CEP库是为Flink的Java API编写的，因此还不支持Scala case类。

作为一种变通方法，您可以将case类转换为普通的Scala类。

还有一个跟踪CEP Scala API开发的JIRA票证。