No Alerts with FinkCEP

[Biplob Biswas]

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.