Hi Biplob, For the 1.4 version, the input of the select function has changed to expect a list of matching events (Map<String, List<IN>> map instead of Map<String, IN> map), as we have added quantifiers.
Also the FIlterFunction has changed to SimpleCondition. The documentation is lagging a bit behind, but it is coming soon. Now for the code I will have to dig into it a bit more. Kostas > On May 26, 2017, at 4:07 PM, Biplob Biswas <revolutioni...@gmail.com> wrote: > > Hello Kostas, > > Thanks for the suggestions. > > I checked and I am getting my events in the partitionedInput stream when i > am printing it but still nothing on the alert side. I checked flink UI for > backpressure and all seems to be normal (I am having at max 1000 events per > second on the kafka topic so don't think backpressure could be a problem, > atleast I expect so) > > Also, I haven't run my test with my test data as a collection but I tried > with this following example and I did get alerts as a result: > > > // CEPTest using collection > > List<MyEvent> inputElements = new ArrayList<>(); > inputElements.add(new MyEvent(1, 'a', 1, 1)); > inputElements.add(new MyEvent(1, 'b', 1, 2)); > inputElements.add(new MyEvent(1, 'b', 2, 2)); > inputElements.add(new MyEvent(1, 'b', 3, 5)); > > Pattern<MyEvent, ?> pattern = Pattern.<MyEvent>begin("a").where(new > FilterFunction<MyEvent>() { > private static final long serialVersionUID = 7219646616484327688L; > > @Override > public boolean filter(MyEvent myEvent) throws Exception { > return myEvent.getPayload() == 'a'; > } > }).followedBy("b").where(new FilterFunction<MyEvent>() { > private static final long serialVersionUID = 7219646616484327688L; > > @Override > public boolean filter(MyEvent myEvent) throws Exception { > return myEvent.getPayload() == 'b'; > } > }).within(Time.seconds(1));//.within(Time.milliseconds(2L)); > > StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); > env.getConfig().setAutoWatermarkInterval(1000); > > DataStream<MyEvent> input = > env.fromCollection(inputElements).assignTimestampsAndWatermarks(new > AscendingTimestampExtractor<MyEvent>() { > private static final long serialVersionUID = -6619787346214245526L; > > @Override > public long extractAscendingTimestamp(MyEvent myEvent) { > return myEvent.getTimestamp(); > } > }); > > PatternStream<MyEvent> patternStream = CEP.pattern(input.keyBy(new > KeySelector<MyEvent, Long>() { > private static final long serialVersionUID = 6928745840509494198L; > > @Override > public Long getKey(MyEvent myEvent) throws Exception { > return myEvent.getId(); > } > }), pattern); > > > patternStream.select(new PatternTimeoutFunction<MyEvent, String>() { > @Override > public String timeout(Map<String, MyEvent> map, long l) throws > Exception { > return map.toString() +" @ "+ l; > } > > private static final long serialVersionUID = 300759199619789416L; > > > }, new PatternSelectFunction<MyEvent, String>() { > > @Override > public String select(Map<String, MyEvent> map) throws Exception { > return map.toString(); > } > > private static final long serialVersionUID = 732172159423132724L; > }).print(); > > > > Also along with that now I upgraded my flink maven project to 1.4-Snapshot > and there seems to be a problem there. > > According to this > <https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/libs/cep.html> > > : > > class MyPatternSelectFunction<IN, OUT> implements PatternSelectFunction<IN, > OUT> { > @Override > public OUT select(Map<String, IN> pattern) { > IN startEvent = pattern.get("start"); > IN endEvent = pattern.get("end"); > return new OUT(startEvent, endEvent); > } > } > > but when I am doing it it expects a list from my side for the events: > > class MyPatternSelectFunction<IN, OUT> implements PatternSelectFunction<IN, > OUT> { > @Override > public OUT select(Map<String, List<IN>> map) throws Exception { > return null; > } > } > > Not really sure what am I doing wrong here, any inputs would be really > helpful. > > Regards, > Biplob > > > > > > -- > View this message in context: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/No-Alerts-with-FinkCEP-tp13333p13341.html > Sent from the Apache Flink User Mailing List archive. mailing list archive at > Nabble.com.