java

6.1 Flink CEP：模式匹配入门

Flink CEP 复杂事件处理入门：Pattern 定义与 begin/next/followedBy 三种跟随模式、where 条件过滤、within 时间窗口约束、PatternSelectFunction 结果提取，以及简单的异常检测模式示例。

Leeting Yan 2026-11-01 3 分钟阅读 1083 字

CEP 是什么

CEP（Complex Event Processing）在流数据中检测模式——不是单条数据的条件判断，而是多条数据按特定顺序和时间关系的组合。

示例：检测「用户在 5 分钟内连续登录失败 3 次」

事件流：login_fail → login_fail → login_fail → login_success
                            ↑
                     CEP 匹配到这个模式 → 输出告警

Pattern API

import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.CEP;

// 定义模式：连续 3 次登录失败
Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("first")
    .where(evt -> evt.status().equals("FAIL"))
    .next("second")
    .where(evt -> evt.status().equals("FAIL"))
    .next("third")
    .where(evt -> evt.status().equals("FAIL"))
    .within(Duration.ofMinutes(5));

三种跟随模式

模式	含义	示例
`next()`	严格连续，中间不能有其他事件	A → B（A 之后紧邻 B）
`followedBy()`	宽松连续，允许中间有其他事件，只取第一条完成的匹配	A → … → B
`followedByAny()`	宽松连续，对每个能完成模式的 B 都产出一条匹配	A → … → B₁、A → … → B₂

// next：失败后必须是失败（中间不能有成功或其他）
Pattern.begin("fail1").where(...).next("fail2").where(...);

// followedBy：失败后可以有其他事件，但最终再失败
Pattern.begin("fail1").where(...).followedBy("fail2").where(...);

where 条件

Pattern.<OrderEvent>begin("large")
    .where(evt -> evt.amount() > 10000)
    .and(evt -> evt.currency().equals("USD"))  // 与条件
    .or(evt -> evt.amount() > 50000);           // 或条件

within 时间约束

Pattern.begin("start")
    .where(...)
    .next("end")
    .where(...)
    .within(Duration.ofMinutes(10));  // 整个模式必须在 10 分钟内完成

超过 within 时间未完成的模式会被丢弃。

匹配结果提取

Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("first")
    .where(evt -> evt.status().equals("FAIL"))
    .next("second")
    .where(evt -> evt.status().equals("FAIL"))
    .next("third")
    .where(evt -> evt.status().equals("FAIL"))
    .within(Duration.ofMinutes(5));

// 本章的 LoginEvent/OrderEvent 都是示例 record，访问器是 userId()/status()
PatternStream<LoginEvent> patternStream = CEP.pattern(loginStream.keyBy(LoginEvent::userId), pattern);

DataStream<Alert> alerts = patternStream
    .select(new PatternSelectFunction<LoginEvent, Alert>() {
        @Override
        public Alert select(Map<String, List<LoginEvent>> pattern) {
            LoginEvent first = pattern.get("first").get(0);
            LoginEvent second = pattern.get("second").get(0);
            LoginEvent third = pattern.get("third").get(0);
            return new Alert("BRUTE_FORCE", first.userId(),
                "3 failed logins in 5 min: " + first.time() + " to " + third.time());
        }
    });

pattern Map 的 key 是 Pattern 中定义的名称（“first”、“second”、“third”），value 是匹配到的事件列表（通常只有一条，除非使用 times()）。

简单模式示例

示例 1：检测大额订单后的小额退款

Pattern.<OrderEvent>begin("large")
    .where(evt -> evt.type().equals("ORDER") && evt.amount() > 10000)
    .followedBy("refund")
    .where(evt -> evt.type().equals("REFUND") && evt.amount() < 100)
    .within(Duration.ofHours(24));

示例 2：检测异常交易序列

Pattern.<TradeEvent>begin("start")
    .where(evt -> evt.symbol().equals("BTC"))
    .next("spike")
    .where(evt -> evt.priceChange() > 0.1)  // 涨 10%
    .followedBy("crash")
    .where(evt -> evt.priceChange() < -0.05)  // 跌 5%
    .within(Duration.ofMinutes(30));

超时处理

模式在 within 时间内未完成，可以输出超时事件。注意超时函数没有单独调用的 select(timeoutFn) 重载——必须和正常匹配函数一起传，结果用 Either 区分，或用 OutputTag 把超时事件侧输出：

PatternStream<LoginEvent> patternStream = CEP.pattern(stream, pattern);

// 方式 1：一次 select 返回 Either<超时, 匹配>
DataStream<Either<Alert, Alert>> results = patternStream.select(
    // 部分匹配但未在 within 内完成
    (PatternTimeoutFunction<LoginEvent, Alert>) (partial, timeoutTimestamp) ->
        new Alert("TIMEOUT", null, "Pattern not completed within time"),
    // 完整匹配
    (PatternSelectFunction<LoginEvent, Alert>) match -> {
        LoginEvent first = match.get("first").get(0);
        return new Alert("BRUTE_FORCE", first.userId(), "3 failed logins in 5 min");
    });

// 方式 2：超时事件走侧输出，主流只保留完整匹配
OutputTag<Alert> timeoutTag = new OutputTag<Alert>("timeouts") {};
SingleOutputStreamOperator<Alert> matches = patternStream.select(
    timeoutTag,
    (PatternTimeoutFunction<LoginEvent, Alert>) (partial, ts) ->
        new Alert("TIMEOUT", null, "Pattern not completed within time"),
    (PatternSelectFunction<LoginEvent, Alert>) match ->
        new Alert("BRUTE_FORCE", match.get("first").get(0).userId(), "3 failed logins"));
DataStream<Alert> timeouts = matches.getSideOutput(timeoutTag);

CEP 的限制

限制	说明
状态大小	每个未完成的模式占用状态内存
within 时间	过长导致状态膨胀
建议先 keyBy	`CEP.pattern(DataStream, pattern)` 能编译运行，但每个并行子任务独立匹配，跨子任务的事件不会被关联；要全局语义就 keyBy，或把并行度设成 1
重复语法有限	只有 `oneOrMore()` / `times(n)` / `timesOrMore(n)`，没有 Drools 那种 `loop` 语法

小结

概念	API
模式定义	`Pattern.begin()`
严格跟随	`.next()`
宽松跟随	`.followedBy()`
条件过滤	`.where()`
时间约束	`.within()`
结果提取	`.select()`
超时处理	`.select()` + `PatternTimeoutFunction`

下一节 6.2 Flink CEP：复杂事件处理。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页