Basic Concepts and Architecture

Flink CEP (Complex Event Processing) is a core component of Apache Flink, specifically designed for processing complex event streams. Built on Flink’s DataStream API, it provides a complete pattern matching framework for real-time pattern detection and analysis on continuously arriving streaming data.

Core Features

  1. Pattern Definition Capability:

    • Supports defining complex event patterns with various constraint conditions
    • Provides time constraints (within), count constraints (times), loop patterns (oneOrMore, timesOrMore), etc.
    • Supports greedy/non-greedy quantifier matching strategies

  2. Event Matching Mechanism:

    • Efficient matching algorithm based on NFA (Non-deterministic Finite Automaton)
    • Supports two matching strategies: strict contiguity and relaxed contiguity
    • Capability to handle out-of-order events (combined with Watermark mechanism)

  3. Result Processing:

    • Matched results can trigger custom operations
    • Supports converting matched event sequences to POJO or Tuple output
    • Rich API for processing matched results (select/process, etc.)

Typical Application Scenarios

  1. Financial Domain:

    • Credit card fraud detection (e.g., multiple large transactions in short time)
    • Abnormal transaction pattern recognition (e.g., money laundering behavior patterns)
    • Real-time risk warning systems

  2. IoT Domain:

    • Device abnormal status monitoring (e.g., temperature continuously rising + pressure sudden drop)
    • Production line fault prediction (abnormal signals after specific operation sequences)
    • Device lifecycle management

  3. Network Security:

    • Intrusion detection (e.g., successful login after multiple login failures)
    • DDoS attack identification (massive requests in short time)
    • Abnormal access pattern analysis

  4. E-commerce Domain:

    • User behavior analysis (browse-add to cart-order pattern recognition)
    • Real-time marketing activity triggering (users meeting specific browsing paths)
    • Abnormal order brushing detection

Technical Advantages

  1. Low Latency, High Throughput: Based on Flink streaming engine, can achieve high throughput processing with millisecond-level latency
  2. Exactly-Once Semantics: Guarantees exactly-once event processing
  3. State Management: Built-in complete state management mechanism, supports large-scale state persistence
  4. Fault Tolerance: Checkpoint-based fault recovery capability
  5. Scalability: Can horizontally scale to process massive event streams

Basic Concepts

Basic Definition

Complex Event Processing (CEP) is an analysis technology based on event streams in dynamic environments. Here events usually represent meaningful state changes. By analyzing relationships between events, using filtering, correlation, aggregation and other technologies, and formulating detection rules based on temporal and aggregation relationships between events, continuously query event streams for event sequences meeting requirements, ultimately analyzing to get more complex compound events.

Feature Definition

CEP’s characteristics are:

  • Goal: Discover some high-level features from ordered simple event streams
  • Input: One or more event streams composed of simple events
  • Processing: Identify connections between simple events, multiple simple events meeting certain rules form complex events
  • Output: Complex events meeting the rules

Function Summary

CEP is used to analyze event streams from different sources that are generated frequently with low latency. CEP helps find meaningful patterns and complex relationships in complex, unrelated event streams, to obtain notifications or organize some behaviors in near real-time or quasi-real-time.

CEP supports pattern matching on streams. According to different patterns, conditions can be continuous or discontinuous. Pattern conditions allow time constraints. When conditions within the range are not met, it can cause pattern matching timeout.

CEP’s main functions include:

  • Generate results as quickly as possible from input stream data
  • Perform aggregation-like calculations based on time on 2 event streams
  • Provide real-time/quasi-real-time alerts and notifications
  • Generate correlation analysis patterns in diverse data sources
  • High throughput, low latency processing

Main Components

Flink provides a dedicated Flink CEP Library, containing the following components: EventStream, Pattern definition, Pattern detection, and Alert generation.

First, developers define pattern conditions on DataStream, then Flink CEP engine performs pattern detection, generating alerts when necessary.

Pattern API

Rules for processing events are called Pattern.

FlinkCEP provides Pattern API for defining complex event rules on input stream data, used to extract event sequences meeting rules.

Patterns are roughly divided into three categories:

  • Individual Patterns: Each single pattern definition that makes up a complex fee—individual patterns
  • Combined Patterns (also called Sequence Patterns): Many individual patterns combined form the entire pattern sequence
  • Group Of Pattern: Using a pattern sequence as a condition nested in individual patterns—becomes a group of patterns

Individual Patterns

Individual patterns include singleton patterns and looping patterns. Singleton patterns accept only one event, while looping patterns can accept multiple events.

Quantifiers

Can append quantifiers to an individual pattern, specifying loop count.

// Match occurring 4 times
start.times(4)
// Match occurring 0 or 4 times
start.times(4).optional
// Match occurring 2, 3 or 4 times
start.times(2,4)
// Match occurring 2, 3 or 4 times, matching as many times as possible
start.times(2,4).greedy
// Match occurring 1 or more times
start.oneOrMore
// Match occurring 0, 2 or more times, matching as many times as possible
start.timesOrMore(2).optional.greedy

Conditions

Each pattern needs to specify trigger conditions, as the judgment basis for whether the pattern accepts events. CEP’s individual patterns mainly set conditions through where, or, until. According to different calling methods, they can be divided into:

  • Simple Conditions: Use where method to filter events by field judgments start.where(event=>event.getName.startsWith("foo"))
  • Combined Conditions: Merge simple conditions, or method represents OR logic, direct combination of where is equivalent to AND
  • Termination Conditions: If using oneOrMore or oneOrMore.optional, recommend using until as termination condition to clean up state
  • Iterative Conditions: Can process all previously accepted events for the pattern, call where, can call ctx.getEventForPattern("name")

Pattern Sequence

Neighbor Patterns

  • Strict Neighbor: All events appear in strict order, with no mismatched events in between, defined by next—for pattern: a next b, event sequence: a c b1 b2 has no match
  • Relaxed Neighbor: Allows mismatched events in between, defined by followedBy—for pattern a followed by b, event sequence: a c b1 b2, matches: a, b1
  • Non-deterministic Relaxed Neighbor: Further relaxes conditions, previously matched events can also be used again, defined by followByAny—for pattern a followByAny b, event sequence: a c b1 b2, matches: ab1, ab2

In addition to the above sequence patterns, can also define that some neighbor relationships should not occur:

  • notNext: Don’t want a certain event to strictly neighbor the previous event
  • notFollowBy: Don’t want a certain event to occur between two events

Additional Notes

  • All pattern sequences must start with begin
  • Pattern sequences cannot end with notFollowedBy
  • not-type patterns cannot be modified by optional
  • Can specify time constraints for patterns, requiring matching within what time period

Pattern Detection

After specifying the pattern sequence to find, can apply it to input stream to detect potential matches. Call CEP.pattern(), given input stream and pattern, get PatternStream

val input:DataStream[Event] = …
val pattern:Pattern[Event,_] = …
val patternStream:PatternStream[Event]=CEP.pattern(input,pattern)

Matched Event Extraction

After creating PatternStream, can apply select or flatSelect methods to extract events from detected event sequences.

select() method needs to input a select function as parameter. Each successfully matched event sequence calls it.

select() receives matched event sequence as Map[String, Iterable[IN]], where key is each pattern’s name, and value is Iterable type of all received events.

def selectFn(pattern : Map[String,Iterable[IN]]):OUT={
  val startEvent = pattern.get("start").get.next
  val endEvent = pattern.get("end").get.next
  OUT(startEvent, endEvent)
}

flatSelect implements similar functionality to Select via PatternFlatSelectFunction. The only difference is flatSelect method can return multiple records. It passes data to downstream through a Collector[OUT] type parameter.

Development Example

// Define event pattern
Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("start")
    .where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent value) {
            return value.getType().equals("fail");
        }
    })
    .next("middle").where(new SimpleCondition<LoginEvent>() {
        @Override
        public boolean filter(LoginEvent value) {
            return value.getType().equals("fail");
        }
    })
    .within(Time.seconds(10));

// Apply pattern on data stream
PatternStream<LoginEvent> patternStream = CEP.pattern(loginEventStream, pattern);

// Process matching results
DataStream<Alert> alerts = patternStream.process(
    new PatternProcessFunction<LoginEvent, Alert>() {
        @Override
        public void processMatch(
            Map<String, List<LoginEvent>> match,
            Context ctx,
            Collector<Alert> out) {
            out.collect(new Alert("Continuous login failure alert"));
        }
    });

Main Concepts

Flink CEP performs complex event processing based on the following core concepts:

Event Stream

Events are basic data units to be processed in the system, usually structured data records with timestamp markers.

Event streams are continuous sequences of these events, with the following characteristics:

  1. Unbounded: Data continuously generated, no predefined endpoint
  2. Ordered: Events usually arrive in time order, but out-of-order may occur
  3. Real-time: Requires low latency processing

State Machine

Flink CEP internally uses optimized Non-deterministic Finite Automaton (NFA) to execute pattern matching:

  1. State Machine Structure:

    • Each pattern step corresponds to a state
    • Transitions triggered by events, based on pattern-defined conditions
    • Can be in multiple states simultaneously (non-deterministic)

  2. Matching Process:

    • Initial state: Waiting for first pattern start
    • Intermediate state: Partial matching in progress
    • Final state: Complete match success

  3. State Management:

    • Use shared state backend to store partial matches
    • Support fault tolerance mechanism to ensure correct matching continues after recovery

Time Processing

CEP supports two time concepts and corresponding processing mechanisms:

  1. Time Types:

    • Event Time: Time when events actually occur, usually embedded in event data
    • Processing Time: System time when Flink processes the event

  2. Time Constraints:

    • Set pattern’s time window through .within() method
    • Example: Event sequence completed within 10 minutes

  3. Out-of-Order Processing:

    • Watermark mechanism: Marks event time progress, allows delayed processing
    • Delay strategy: Can set maximum allowed delay time
    • Side output: Output timeout events to side stream for special processing

Example Time Processing:

Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
    .where(...)
    .next("middle")
    .where(...)
    .within(Time.seconds(30));  // Complete matching within 30 seconds

CEP.pattern(stream, pattern)
    .inEventTime()  // Use event time
    .withLateDataOutputTag(outputTag)  // Process late data
    .select(...);
  • Real-time: Flink itself is a real-time stream processing framework, and CEP enables it to process complex event patterns, allowing users to detect and respond in real-time
  • Scalability: Flink CEP is based on distributed architecture, can process high-throughput data streams and run on large-scale clusters
  • Flexibility: Users can define various complex event patterns through simple API to meet various business requirements