flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Kostas Kloudas <k.klou...@data-artisans.com>
Subject Re: No Alerts with FinkCEP
Date Fri, 26 May 2017 14:36:52 GMT
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 <revolutionisme@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&lt;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.


Mime
View raw message