Apach Flink CEP "At Least" Condition

Question

I am trying to create a CEP pattern that matches "at least" occurrence. Modifying the example code:

middle.oneOrMore().where(new IterativeCondition<SubEvent>() {
    @Override
    public boolean filter(SubEvent value, Context<SubEvent> ctx) throws Exception {
        if (!value.getName().startsWith("foo")) {
            return false;
        }

        double sum = value.getPrice();
        for (Event event : ctx.getEventsForPattern("middle")) {
            sum += event.getPrice();
        }
        return Double.compare(sum, 5.0) < 0;
    }
});

into

middle.oneOrMore().where(new IterativeCondition<SubEvent>() {
    @Override
    public boolean filter(SubEvent value, Context<SubEvent> ctx) throws Exception {
        if (!value.getName().startsWith("foo")) {
            return false;
        }

        long count = 0;
        for (Event event : ctx.getEventsForPattern("start")) {
            count = count + 1; 
        }
        return count >= MIN_COUNT;
    }
});

does not solve my problem because the condition will keep failing and never be able to contribute to count.

I found in https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/libs/cep.html that there are

// expecting 4 occurrences
 start.times(4);

 // expecting 0 or 4 occurrences
 start.times(4).optional();

 // expecting 1 or more occurrences
 start.oneOrMore();

 // expecting 0 or more occurrences
 start.oneOrMore().optional();

Is there anything like start.atLeast(5) exist?

Maybe use next() to combine times(4) with oneOrMore()? Haven't tried it, but seems like it should work. — David Anderson
Thanks, I played with times(n).next("pattern2").oneOrMore(), bu it didn't seem to do what I need. I also played with oneOrMore().next("pattern2").time(n), and it produced closer result. I will continue investigating this issue and share my findings later. — Wei Ma

Jay Vignesh Jay Vignesh · Accepted Answer · 2017-07-12T07:01:56

Check this link for the solution. It checks for the pattern to occur 5 times. You can modify it in times(number_of_times)

[https://stackguides.com/questions/45033109/flink-complex-event-processing/45048866]

Apach Flink CEP "At Least" Condition

2 Answers