Is there a work-around to handle multiple “temporal constraints” in Flink CEP?

Question

As stated in CEP document (https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/libs/cep.html) that only one temporal constraint is allowed in a pattern sequence, I'm struggling to find out a way to handle a business case that contains 2 temporal constraints.

I need to monitor some business events and alert on the events that meet the following rules:

a new account is registered
the account gets authenticated in 5 minutes after registration
the account completes at least 2 transactions which transaction amount is greater than 1000.00 in next 1 hour.

And the code is something like this:

Pattern<Event, ?> pattern = Pattern.<Event>begin("register").where(new SimpleCondition<Event>() {
    @Override
    public boolean filter<Event value> throws Exception {
        return (value.getEventType() == EventType.REGISTER);
    }
}).followedBy("authentication").where(new SimpleCondition<Event>() {
    @Override
    public boolean filter<Event value> throws Exception {
        return (value.getEventType() == EventType.AUTHENTICATION);
    }
}).where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context<Event> ctx) throws Exception {
        for (Event event : ctx.getEventsForPattern("register")) {
            if (value.getEventTime() - event.getEventTime() <= 1000 * 60 * 5) {
                return true;
            }
        }
        return false;
    }
}).followedBy("transaction").where(new SimpleCondition<Event>() {
    @Override
    public boolean filter<Event value> throws Exception {
        return (value.getEventType() == EventType.TRANSACTION && value.getAmount() > 1000.00);
    }
}).where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context<Event> ctx) throws Exception {
        for (Event event : ctx.getEventsForPattern("authentication")) {
            if (value.getEventTime() - event.getEventTime() <= 1000 * 60 * 60) {
                return true;
            }
        }
        return false;
    }
}).timesOrMore(2);

You can see that I use 2 IterativeConditions to handle the temporal constraints. Is there a better way to make the code more concise?

Dawid Wysakowicz Dawid Wysakowicz · Accepted Answer · 2018-08-09T06:53:21

As you said you can apply only one time constraint to whole pattern right now in CEP library. What you could do though is to split you pattern into 2 sub patterns. First apply pattern that will look for REGISTER -> AUTHENTICATE and generate complex event out of those (let's name it REGISTER_AUTHENTICATED). And then use it in the subsequent pattern REGISTER_AUTHENTICATED -> 2* TRANSACTIONS.

Then you can apply two time constraints to both of those patterns.

Is there a work-around to handle multiple “temporal constraints” in Flink CEP?

1 Answers