@hegdemahendra , You need to set the time characteristic of the stream for it to work. For example, try setting it to processing time, as shown below: DataStream<String> matechedStream = patternStream
.inProcessingTime()
.process(new PatternProcessFunction<String, String>() {
@Override
public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
collector.collect(map.get("start").toString());
}
}
); Cheers, André
... View more
