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 13:34:58 GMT
One additional comment, from your code it seems you are using Flink 1.2.
It would be worth upgrading to 1.3. The updated CEP library includes a lot of 
new features and bugfixes.

Cheers,
Kostas

> On May 26, 2017, at 3:33 PM, Kostas Kloudas <k.kloudas@data-artisans.com> wrote:
> 
> Hi Biplob,
> 
> From a first scan of the code I cannot find sth fishy.
> 
> You are working on ProcessingTime, given that you do not 
> provide any time characteristic specification, right?
> 
> In this case, if you print your partitionedInput stream, do you 
> see elements flowing as expected?
> 
> If elements are flowing normally, is any back pressure created? 
> Or you keep on reading records from kafka uninterrupted? 
> I am asking to see if the CEP operator is doing sth that blocks the 
> pipeline or it just discards the elements.
> 
> It could be also worth trying to add a source with artificial elements env.fromCollection(…)

> to see if in this case everything works normally.
> 
> Kostas
> 
>> On May 26, 2017, at 2:25 PM, Biplob Biswas <revolutionisme@gmail.com> wrote:
>> 
>> Hi,
>> 
>> I just started exploring Flink CEP a day back and I thought I can use it to
>> make a simple event processor. For that I looked into the CEP examples by
>> Till and some other articles. 
>> 
>> Now I have 2 questions which i would like to ask:
>> 
>> *Part 1:*
>> 
>> I came up with the following piece of code, but this is not working as
>> expected.
>> 
>> ///**************** Main ******************///
>> 
>> 
>> FlinkKafkaConsumer010<String> consumer = new FlinkKafkaConsumer010<>(
>>           "testTopic",
>>           new SimpleStringSchema(),
>>           props);
>> 
>>   DataStream<String> input = env.addSource(consumer);
>>   LOG.info("About to process events");
>>   DataStream<ReadEventType> events =
>>           input
>>                   //.map(s -> s.f1)
>>                   .map(new MapStringToRRE())
>>                   .filter(Objects::nonNull);
>> 
>>   //events.print();
>> 
>>   DataStream<ReadEventType> partitionedInput = events
>>           .keyBy((KeySelector<ReadEventType, String>) value ->
>> value.getRawTransactionItem().getChargedAccount());
>> 
>>   Pattern<ReadEventType, ?> pattern =
>> Pattern.<ReadEventType>begin("first")
>>           .where(event -> event.getFormat() == FormatType.FILE)
>>           .followedBy("second")
>>           .where(event -> event.getFormat() == FormatType.FILE)
>>           .within(Time.seconds(1));
>> 
>>   PatternStream<ReadEventType> patternStream =
>> CEP.pattern(partitionedInput, pattern);
>> 
>>   DataStream<String> alerts =
>> patternStream.select((PatternSelectFunction<ReadEventType, String>)
>> CEPForBAMRRE::createAlert);
>> 
>>   alerts.print();
>> 
>>   env.execute("CEP monitoring job");
>> }
>> 
>> 
>> ///*********** Alert Function returning just concat of txn id
>> ***************///
>> 
>> private static String createAlert(Map<String, ReadEventType> pattern) {
>>   return pattern.get("first").getTransactionItem().getUid() + " " +
>>           pattern.get("second").getTransactionItem().getUid();
>> }
>> 
>> ///******************* properties for kafka **************///
>> 
>> private static Properties getDefaultProperties(Properties prop){
>>   prop.put("group.id", "FlinkCEP");
>>   prop.put("bootstrap.servers", BOOTSTRAP_SERVERS);
>>   prop.put("zookeeper.connect", ZKEEPER);
>>   prop.put("auto.offset.reset", "earliest");
>>   return prop;
>> }
>> 
>> 
>> As my kafka topic only sends me events with formattype = FILE, I was
>> expecting to see multiple alerts being raised. But thats not the case, i am
>> not getting any alert at the moment.
>> 
>> Can anyone point out what am I doing wrong? 
>> 
>> PART 2: 
>> 
>> Also, my main aim for using CEP is to read from different topics and raise
>> alert if a second event is *not* followed by a first event within a given
>> time interval. How can I achieve it with FlinkCEP? for now I can only see
>> that if 2 events follow within a time interval an alert should be raised. 
>> 
>> 
>> Thanks & Regards,
>> Biplob
>> 
>> 
>> 
>> --
>> View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/No-Alerts-with-FinkCEP-tp13333.html
>> Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
> 


Mime
View raw message