Learn Apache Flink - Complex Event Processing (CEP)

#Introduction

So far we've processed events one by one or aggregated them in windows. Episode 10 changes the perspective: what if what we're looking for is a sequence of events — a suspicious transaction followed by another within a minute, or a series of logs signaling a cascading failure? This is the realm of Complex Event Processing (CEP).

Flink provides a complete CEP library with the Pattern API. We'll learn to define patterns, match them against an event stream, apply them to fraud detection and anomaly detection, and handle timed patterns and pattern states. This is one of the capabilities that makes Flink stand out in its class.

#CEP Concepts in Flink

#Patterns as Descriptions of Sequences

CEP works by defining patterns — descriptions of the sequence of events you want to detect. Each pattern consists of several states: begin, next, followedBy, with conditions that filter which events match. When the sequence of events in the stream matches the pattern, Flink produces a match.

begin("a") → next("b") → followedBy("c") → within(1 menit)

The sequence above means: event a, then b immediately after, then c somewhere after that, all within a one-minute span.

#The Pattern API

The Pattern API is expressed as a chain of method calls. Let's look at a pattern definition for detecting two consecutive large transactions:

import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.windowing.time.Time;
 
Pattern<Transaction, ?> pattern = Pattern.<Transaction>begin("t1")
    .where(new SimpleCondition<Transaction>() {
        @Override
        public boolean filter(Transaction t) {
            return t.getAmount() > 500000;
        }
    })
    .next("t2")
    .where(new SimpleCondition<Transaction>() {
        @Override
        public boolean filter(Transaction t) {
            return t.getAmount() > 500000;
        }
    })
    .within(Time.minutes(1));

.next("t2") demands that event t2 appears immediately after t1, while within(Time.minutes(1)) bounds the whole pattern to one minute. Understand that next is strict: no other event in between.

#Using CEP on a Stream

#Wrapping a Stream with CEP.pattern

A pattern is useless until it's attached to a stream. Use CEP.pattern and then process each match:

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
 
DataStream<Transaction> keyed = transactions.keyBy(Transaction::getAccount);
 
PatternStream<Transaction> patternStream = CEP.pattern(keyed, pattern);
 
DataStream<Alert> alerts = patternStream
    .process(new PatternProcessFunction<Transaction, Alert>() {
        @Override
        public void processMatch(
                Map<String, List<Transaction>> match,
                Context ctx,
                Collector<Alert> out) {
            Transaction t1 = match.get("t1").get(0);
            Transaction t2 = match.get("t2").get(0);
            out.collect(new Alert(t1.getAccount(), t1.getAmount(), t2.getAmount()));
        }
    });

match.get("t1") retrieves all events that match the t1 state. From here, your business logic (creating alerts, writing to a sink) just needs to be connected.

#The Difference Between next, followedBy, and Quantifiers

• next — the next event must be directly consecutive.
• followedBy — the next event appears afterwards, other events may interleave.
• times(2) and oneOrMore() — repeat a state several times.

Pattern.<Event>begin("mulai")
    .where(e -> e.getType().equals("ping"))
    .times(2)
    .next("akhir")
    .where(e -> e.getType().equals("pong"));

#Use Cases: Fraud Detection and Anomaly Detection

#A Fraud Schema with a Double Threshold

The classic case: two large transactions from the same account in a short time. The pattern above already captures this. Strengthen it by adding per-account conditions, for example comparing against a historical average stored in state.

#Anomalies and Workflow Automation

CEP isn't just for fraud. Other examples:

• Anomaly detection: an error rate spike followed by a health check drop within a few minutes.
• Workflow automation: the login → add_to_cart → checkout checkout sequence read for conversions.
• Failure cascades: a series of services going down that signals a major incident.

./bin/flink run -d target/cep-job.jar

The ./bin/flink run -d command submits the CEP job to the cluster. In production, match results are usually written to Kafka or an alerting system.

#Timed Patterns and Pattern States

#Time Bounds and State Management

A pattern with .within() makes Flink store temporary state for each stream key until the time bound is reached or the pattern is satisfied. This means CEP uses memory — monitor the state size so it doesn't balloon. Use patterns that are as narrow as possible and set a timeout with a TimeoutHandler to clean up partial matches.

DataStream<Alert> alerts = patternStream
    .process(new PatternTimeoutFunction<Transaction, Alert>() {
        @Override
        public Alert onTimeout(
                Map<String, List<Transaction>> partial,
                long ts,
                Context ctx) {
            return new Alert("timeout", partial.size());
        }
    })
    .setParallelism(1);

PatternTimeoutFunction is called when a pattern doesn't complete within the within bound — giving you visibility over dangling patterns as well as the chance to clean up state.

#Conclusion

Episode 10 introduced Flink CEP: defining patterns with the Pattern API, attaching them to streams with CEP.pattern, processing matches with PatternProcessFunction, and applying those patterns to fraud detection, anomaly detection, and workflow automation. You also understood timed patterns and pattern state management.

The key takeaways:

• CEP detects sequences of events, not just individual events.
• next demands strict ordering, followedBy relaxes it, times repeats a state.
• within bounds a pattern to a time range and is the limit for state cleanup.
• CEP stores state per key — design patterns as narrowly as possible.
• Main use cases: fraud detection, anomaly detection, and workflow automation.

In the next episode, episode 11, we'll discuss job configuration & deployment — assembling job JAR packages with dependency shading, the standalone, YARN, and Kubernetes deployment modes, JobManager and TaskManager resource configuration, and managing the job lifecycle with the CLI and web UI. You'll ship jobs from your laptop all the way to a production cluster.