Definition
- Flink CEP – A library on top of the DataStream API for detecting complex patterns in event streams.
- Pattern builder – Fluent API where
Pattern.begin("login").next("failedPayment").within(Time.minutes(5))defines stages with conditions. - Contiguity –
next()for strict (immediate),followedBy()for relaxed (gaps allowed),followedByAny()for non-deterministic matching. - NFA execution – The pattern compiles into a non-deterministic finite automaton tracked in keyed state per partition key.
- Time constraint –
within()bounds the pattern so partial matches that exceed the window are discarded.
The National Bank of India's security team has a rule: "if a customer logs in, then has a failed payment, then logs out, all within five minutes, that is suspicious." Flink CEP is like a watchful guard who knows that exact story and watches every customer's actions in order. The guard does not react to any single action. But when the three actions happen in that precise order inside five minutes, the alarm rings.
Understanding the Concepts
What is Flink CEP?
Flink CEP is a library built on the DataStream API for detecting patterns in event streams. You describe a pattern as a chain of named stages, each with a condition, and Flink compiles it into a stateful automaton.
package com.bank.flink;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.util.List;
import java.util.Map;
public class FraudPatternDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
KeyedStream<BankEvent, String> events = env
.addSource(new KafkaBankEventSource())
.keyBy(BankEvent::getAccountId);
// Define the fraud pattern
Pattern<BankEvent, ?> fraudPattern = Pattern
.<BankEvent>begin("login")
.where(new SimpleCondition<BankEvent>() {
public boolean filter(BankEvent event) {
return "LOGIN".equals(event.getType());
}
})
.next("failedPayment")
.where(new SimpleCondition<BankEvent>() {
public boolean filter(BankEvent event) {
return "PAYMENT_FAILED".equals(event.getType());
}
})
.next("logout")
.where(new SimpleCondition<BankEvent>() {
public boolean filter(BankEvent event) {
return "LOGOUT".equals(event.getType());
}
})
.within(Time.minutes(5));
PatternStream<BankEvent> patternStream = CEP.pattern(events, fraudPattern);
DataStream<String> alerts = patternStream.select(
new PatternSelectFunction<BankEvent, String>() {
public String select(Map<String, List<BankEvent>> match) {
BankEvent login = match.get("login").get(0);
BankEvent failed = match.get("failedPayment").get(0);
return "FRAUD ALERT: " + login.getAccountId()
+ " login→failedPayment→logout in 5 min";
}
}
);
alerts.print();
env.execute("NBI-Fraud-CEP");
}
}
What are contiguity operators?
Contiguity operators define how strictly events must follow each other.
// Strict contiguity: next event must immediately match
Pattern.<BankEvent>begin("A").next("B")
// Events: A, B → match
// Events: A, X, B → no match (X is in between)
// Relaxed contiguity: allows unrelated events between
Pattern.<BankEvent>begin("A").followedBy("B")
// Events: A, X, B → match (X is skipped)
// Events: A, X, Y, B → match
// Non-deterministic relaxed: matches all possible Bs
Pattern.<BankEvent>begin("A").followedByAny("B")
// Events: A, B1, B2 → two matches: (A,B1) and (A,B2)
Contiguity comparison:
A X B
next(): ✗ (X blocks the match)
followedBy(): ✓ (X is skipped, first B matches)
followedByAny(): ✓ (all Bs match independently)
What do quantifiers do?
Quantifiers control how many events a stage can match.
// One or more failed payments followed by a logout
Pattern.<BankEvent>begin("login")
.where(e -> "LOGIN".equals(e.getType()))
.followedBy("failures")
.where(e -> "PAYMENT_FAILED".equals(e.getType()))
.oneOrMore() // one or more failures
.followedBy("logout")
.where(e -> "LOGOUT".equals(e.getType()))
.within(Time.minutes(10));
// Other quantifiers:
// .times(3) exactly 3 events
// .times(2, 5) between 2 and 5 events
// .optional() zero or one event
// .greedy() match as many as possible (default)
Why is the within() clause essential?
The within() clause bounds the pattern to a time window. Without it, partial matches would be kept in state indefinitely, causing unbounded memory growth.
// CORRECT: bounded pattern
Pattern.<BankEvent>begin("login")
.next("failed")
.within(Time.minutes(5));
// Partial match discarded after 5 minutes → bounded state
// WRONG: unbounded pattern
Pattern.<BankEvent>begin("login")
.next("failed");
// Partial match kept forever if 'failed' never arrives → memory leak
How does the NFA execution work?
Flink compiles the pattern into a non-deterministic finite automaton (NFA) maintained in keyed state. As events arrive, the automaton advances partial matches. When a full match completes, it emits the result.
NFA for Pattern: login → failedPayment → logout (within 5 min)
State 0 State 1 State 2 State 3
(start) ──login──▶ (login ──failed──▶ (login+ ──logout──▶ (MATCH!)
matched) failed)
│ │
│ timeout 5 min │ timeout 5 min
▼ ▼
(discard) (discard)
How do you handle timed-out partial matches?
Timed-out partial matches can be captured via a timeout output tag, useful when you want to know a sequence started but never completed.
package com.bank.flink;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.util.List;
import java.util.Map;
public class TimeoutDemo {
static final OutputTag<String> timeoutTag = new OutputTag<>("timeout") {};
public static void processFraudPattern(
SingleOutputStreamOperator<BankEvent> events,
Pattern<BankEvent, ?> pattern) {
SingleOutputStreamOperator<String> result = CEP.pattern(events, pattern)
.flatSelect(
timeoutTag,
// Timeout handler: sequence started but never completed
(Map<String, List<BankEvent>> partial, long timestamp, Collector<String> out) -> {
out.collect("WEAK SIGNAL: login but no logout within window for "
+ partial.get("login").get(0).getAccountId());
},
// Match handler: full sequence completed
(Map<String, List<BankEvent>> match, Collector<String> out) -> {
out.collect("FRAUD ALERT: " + match.get("login").get(0).getAccountId());
}
);
result.print("matches");
result.getSideOutput(timeoutTag).print("timeouts");
}
}
How does CEP compare to MATCH_RECOGNIZE?
| Feature | Flink CEP | MATCH_RECOGNIZE |
|---|---|---|
| API | Java/Scala builder | SQL |
| Where it sits | DataStream API | Flink SQL |
| Dynamic patterns | Slightly more flexible | Statically compiled |
| Readability | Code-heavy | Declarative SQL |
| Timeout handling | Built-in side output | Not directly supported |
| Best for | Java developers | SQL-fluent analysts |
What should a senior engineer know?
- CEP runs on event time with watermarks for deterministic, auditable matching.
- Strict contiguity (next) is cheaper than relaxed (followedBy) because it keeps fewer partial matches.
- For dynamic patterns that change at runtime, consider a KeyedProcessFunction instead.
- CEP state grows with the number of simultaneous partial matches, so bound patterns with within().
- Test patterns with small datasets to verify they match the expected sequences before deploying.
// Senior tip: monitor CEP state size
// High partial-match count → state growing → add within() or tighten contiguity
// Flink metrics: numPartialMatches per key should be bounded
Visual Overview
CEP Pattern Builder Flow
Pattern.begin("login")
│
│ .where(type == LOGIN)
▼
.next("failedPayment")
│
│ .where(type == PAYMENT_FAILED)
▼
.next("logout")
│
│ .where(type == LOGOUT)
▼
.within(Time.minutes(5))
│
▼
CEP.pattern(keyedStream, pattern)
│
▼
PatternStream
│
┌────┴────┐
▼ ▼
select() timeout()
(alerts) (weak signals)
Contiguity Operators Event stream: LOGIN → VIEW → PAYMENT_FAILED → LOGOUT next(): LOGIN ──▶ VIEW (not PAYMENT_FAILED) → NO MATCH followedBy(): LOGIN ──▶ (skip VIEW) ──▶ PAYMENT_FAILED → MATCH next() requires the IMMEDIATE next event to match. followedBy() allows unrelated events in between.
Key Takeaway
Flink CEP is a library on top of the DataStream API for detecting complex patterns in event streams. You describe a pattern with a fluent builder, chaining stages like login, failedPayment, and logout with conditions and contiguity operators, bounded by a within() time constraint. Flink compiles the pattern into a non-deterministic finite automaton tracked in keyed state, and emits matched events through a PatternSelectFunction.
The contiguity operators are the expressive core. next() requires strict contiguity where the immediate next event must match, followedBy() allows gaps, and quantifiers like oneOrMore() let a stage match multiple events. The within() clause is essential to bound partial-match state and prevent memory leaks. Timed-out partial matches can be captured through a side output for weaker signals.
At the National Bank of India, our fraud team used exactly the classic signal: login, then failed payment, then logout for the same account within five minutes. We expressed it as a CEP pattern keyed by account with strict contiguity and a five-minute window. Full matches went to the alert topic, and timed-out partial matches went to a separate case-management stream as weaker signals. Because CEP runs on event time with watermarks, the same events always produced the same matches, which the audit team required.
Remember one sentence: Flink CEP detects ordered event patterns using a fluent Pattern builder compiled into a keyed NFA, where the within() time constraint keeps state bounded and contiguity operators control how strictly events must follow each other.
---