Flink CEP pattern design question

Posted by RayL on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Flink-CEP-pattern-design-question-tp26809.html

Currently I'm designing a CEP pattern to satisfy our business needs.
Basically, there's two events let's call it a and b.
Both a and b can have zero or multiple entries in the log.
For input {a,b1,b2}, I want to get the output of {a,b1,b2}
For input {b1,b2}, I want to get the output of {b1,b2} after a timeout
period(10 seconds)
For input {a1,a2}, I want to get the output of {a1,a2} after a timeout
period(10 seconds)

Currently, my code looks like this.
val pattern = Pattern
                        .begin[Event]("start")
                        .where(_._.getName == "a")
           .oneOrMore.optional
                        .followedBy("end")
                        .where(_._.getName == "b")
                        .oneOrMore.optional
           .within(Time.seconds(10))
For input {a,b1,b2}, I got the output of {a1} {a1,b1} {a,b1,b2} {b1,b2} {b2}
For input {b1,b2}, I got the output of {b1,b2} {b1} {b2}
For input {a1,a2}, I got the output of {a1,a2} {a1} {a2}

Also tried SKIP_PAST_LAST_EVENT policy, and didn't work.

Please advice me how to design this pattern to match my needs.




--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/