如果你用 Flink 做实时计算那你一定绕不开两个东西一个是 FlinkSQL把一堆流数据当成一张可以不停查询的表另一个是 CEPComplex Event Processing复杂事件处理专门用来从事件流里挑出那些“按特定顺序发生”的组合。很多刚接触 Flink 的同学往往把这两个概念分开学SQL 会写了CEP 的模式也能跑通但真到了业务场景里却搞不清楚到底该用哪个、能不能一起用。这篇内容是我自己学到 Flink 第五阶段时的实战总结主要围绕 FlinkSQL 和 CEP 的配合使用展开适合正在做实时数仓、实时风控、用户行为分析或者只是想把手头 Flink 作业写得更有章法的开发者参考。我先把结论摆出来FlinkSQL 擅长的是“聚合、过滤、关联、窗口统计”这类常规数据处理它用一种动态表Dynamic Table的机制把无限流转化成有限查询CEP 擅长的则是“事件的先后顺序”和“组合关系”比如 10 分钟内同一个用户下单后没有支付、又连续点击了三次退款这种模式用纯 SQL 写起来非常别扭甚至写不出来但用 CEP 的 Pattern API 或模式子句就会很顺。弄清楚两者的边界和协作方式你的实时任务会清晰很多排查问题的时候也不会再一头雾水。1. 先看整体FlinkSQL 和 CEP 各自解决什么问题1.1 动态表模型把流变成能查的表FlinkSQL 的核心思想是动态表。普通的数据库表是静态的数据存在磁盘上你查询时它不会自己变而 Flink 的动态表不一样流里的每一条数据进来都会触发这张表的一次更新。你可以简单理解成这张表是一个“一直在自动刷新”的视图刷新频率取决于上游数据到达的速率。这个模型带来了两个很实用的能力。第一个是连续查询Continuous Query你写一条 SELECT 语句它不会跑一次就结束而是持续运行每来一条数据就重新计算一次结果。第二个是结果会随着时间推移发生变化可能出现追加Append模式也可能出现更新Update模式。这也是为什么 FlinkSQL 在写出到 Kafka 或者数据库时要区分连接器是 append-only 还是 upsert 模式——因为这决定了下游表的存储引擎能不能承载更新操作。很多同学在实操时遇到“写不进去”“报错说表不支持更新”的问题根源往往就在这里。动态表的三种流转换类型值得记一下。Append 流只有插入操作适合日志类数据。Retract 流有插入和删除两种消息逻辑上处理起来更复杂但兼容性最好。Upsert 流通过 key 来区分是插入还是更新适合操作用户维表或者状态数据。理解了这三种流你回头看 FlinkSQL 作业在 Checkpoint 里的状态构成、以及遇到数据回撤retract时为什么下游会出现 DELETE 消息就不会觉得奇怪了。1.2 CEP 的定位别用 SQL 硬写状态机CEP 跟 SQL 最大的区别在于它处理的是“事件形状”而不是“数据量”。SQL 的聚合可以告诉你“这个用户今天下过几单”但很难告诉你“这个用户是先加购物车、再下单、最后又取消了并且整个过程在 5 分钟内完成”。后者是一种典型的事件模式需要记录事件的顺序和间隔这就超出了传统 SQL 的表达范围。有人可能会说那我用多个 SQL 加状态算子也能实现可以但非常痛苦。你需要自己维护一堆 ValueState / ListState自己记录每个事件到达的时间戳自己写判断逻辑判断先后顺序。而 CEP 把这些事情封装成了模式语言——你只需要声明“第一个事件是什么第二个事件是什么它们之间相隔多久”剩下的事件缓存、状态管理、超时触发框架全都帮你处理。所以在规划实时任务时我一般会这么分如果业务是一个“数字指标”的需求比如每分钟交易额、每小时的活跃用户数直接上 FlinkSQL如果业务是一个“行为组合判断”的需求比如连续失败三次触发告警、登录后 5 分钟未支付自动关单优先考虑 CEP。两者不是替代关系而是互相配合很多生产任务里是先 SQL 做清洗和聚合再 CEP 做模式识别。2. FlinkSQL 实操核心时间属性、窗口和维表关联2.1 时间属性createTime 还是 procTime到底怎么选FlinkSQL 里的时间属性是很多新手第一个踩坑的地方。表上有两种时间处理时间Processing Time和事件时间Event Time。处理时间就是算子所在机器的当前时间取的是本地时钟简单但不可靠因为只要上游有延迟或者网络抖动处理结果就会跟“真实发生时间”产生偏差。事件时间是业务数据里自带的时间字段比如订单创建时间、日志产生时间更能反映业务实际发生顺序但需要你配合水位线Watermark机制来处理乱序和延迟数据。在声明事件时间时有一点容易忽略Flink 要求事件时间字段必须是 TIMESTAMP(3) 类型而且必须是表中的“逻辑字段”不能直接拿一个 BIGINT 的毫秒时间戳来用。你需要先转换比如TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000))才能顺利声明为事件时间。我实际项目里踩过一个坑上游数据里的时间字段有时是字符串有时是 BIGINT直接在 WITH 子句里声明 rowtime 就报错。后来统一在源表 SQL 里写了一个 computed column 做转换把转换后的字段单独取一个名字再把这个新字段声明为事件时间问题就解决了。类似这种“数据质量参差不齐”的情况在真实生产环境里非常常见建议你在建表时就把清洗逻辑放在 SQL 里而不是依赖上游改造成本。CREATE TABLE order_log ( order_id STRING, user_id STRING, amount DOUBLE, ts BIGINT, rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000)), WATERMARK FOR rowtime AS rowtime - INTERVAL 5 SECOND ) WITH ( connector kafka, topic order_log, properties.bootstrap.servers localhost:9092, format json, scan.startup.mode earliest-offset );看到WATERMARK FOR rowtime AS rowtime - INTERVAL 5 SECOND这一行了吗它的意思是允许事件最多迟到 5 秒超过这个界限就视为“迟到的数据”默认会被丢弃或者发到侧输出流。严格来说这只是一个基于经验值的设定具体设多少要根据你的业务容忍度来定。比如实时大屏可以容忍延迟换取准确性那水位线可以放宽如果是风控场景迟到的数据可能直接造成漏判那么水位线就要设置得比较紧同时配合侧输出流单独处理迟到数据。2.2 窗口聚合翻滚窗口、滑动窗口和累计窗口怎么用窗口是 FlinkSQL 用来做“无界流有界化”的关键算子。三种窗口的差异很简单。TUMBLE翻滚窗口把数据按照固定时间长度切成互不重叠的窗口每个数据只属于一个窗口适合做每分钟独立统计。HOP滑动窗口有两个参数窗口长度和滑动步长数据可能同时属于多个窗口适合做近段趋势分析比如“近 5 分钟的累计交易量每 30 秒刷新一次”。SESSION会话窗口没有固定长度而是按照活跃间隙划分适合统计连续性行为比如用户连续访问多长时间后离开。在 SQL 里写窗口一般有两种方式。一种是在 GROUP BY 子句里直接使用窗口函数比如GROUP BY TUMBLE(rowtime, INTERVAL 1 MINUTE)。另一种是用窗口表值函数Window TVF比如SELECT window_start, window_end, user_id, SUM(amount) AS total_amount FROM TABLE( TUMBLE(TABLE order_log, DESCRIPTOR(rowtime), INTERVAL 1 MINUTE) ) GROUP BY window_start, window_end, user_id;窗口表值函数是相对新一些的写法比旧版 GROUP BY 窗口更灵活因为你可以在窗口上做更复杂的计算包括多级聚合。我推荐在生产环境优先使用 Window TVF它更接近标准 SQL 的语义后续扩展也方便。窗口聚合的结果写出去之前最好设置一个合适的空闲超时Idle Timeout否则某个上游分区长期没数据窗口会被卡住不触发。这个问题在 Kafka 多分区场景下特别容易遇到解决的思路是给源表的水位线加上WATERMARK FOR rowtime AS rowtime - INTERVAL 5 SECOND的同时再设置table.exec.source.idle-timeout 1 min让长时间静默的源分区自动标记为空闲。2.3 维表关联Lookup Join 和普通 Join 的区别实时计算里经常要把流里的用户 id 关联上用户姓名、地区等静态信息。FlinkSQL 支持两种方式。一种是双流 Join两张都是动态表数据到达顺序不确定需要靠状态缓存一段时间。另一种是 Lookup Join一张是实时流另一张是存储在外部系统里的维表比如 MySQL、HBase每条主流数据来到时去外部系统查一次。第二种在实时数仓里用得非常频繁因为维表数据量不大、变化不频繁用外部存储查询完全够用还能避免把维表全量加载到状态里。Lookup Join 的 SQL 写起来很简单重点是FOR SYSTEM_TIME AS OF这个语法它表示“查询这个时刻的维表快照”。例如SELECT o.order_id, o.user_id, u.user_name, o.amount FROM order_log o LEFT JOIN user_dim FOR SYSTEM_TIME AS OF o.rowtime AS u ON o.user_id u.user_id;实际操作中有两个坑。第一个是维表连接器要设置缓存比如lookup.cache.max-rows 1000、lookup.cache.ttl 30s否则每条数据都去打 MySQL延迟高不说数据库也容易被压垮。第二个是维表更新问题维表一般不是实时变的你在外部库里改了数据Flink 这边可能还在用旧缓存这时要结合业务需求调整缓存 TTL不能盲目求快。对于需要秒级感知维表更新的业务可以考虑直接把维表做成 Kafka 主题用普通 Join 来做而不是 Lookup Join。3. CEP 实操核心模式匹配从入门到能上生产3.1 Pattern API用 begin、next、followedBy 描述你的业务CEP 的核心是 Pattern API。你可以把它理解成一种“正则表达式”只不过匹配的对象不是字符而是事件流。begin(start)模式的起点匹配第一个事件。next(pattern)严格连续中间不能插入其他事件。followedBy(pattern)松散连续允许中间出现不相关的事件。followedByAny(pattern)更松散的连续允许多个模式实例重叠。within(time)限定整个模式在时间范围内完成超时就不算。这三个连接词的区别是 CEP 新手最头疼的地方。举个具体的例子假设你要识别“下单后 5 分钟内支付”的行为。使用next时必须是订单事件后紧跟着支付事件中间不能有浏览、加购等其他操作使用followedBy时订单后面隔了几个其他事件再支付也算匹配成功。绝大多数业务场景用的是followedBy因为真实用户行为里总会夹着一些其他事件。Pattern API 代码写起来一般是这样的PatternOrderEvent, OrderEvent pattern Pattern.OrderEventbegin(start) .where(new SimpleConditionOrderEvent() { Override public boolean filter(OrderEvent event) throws Exception { return event.getEventType().equals(CREATE); } }) .followedBy(middle) .where(new SimpleConditionOrderEvent() { Override public boolean filter(OrderEvent event) throws Exception { return event.getEventType().equals(PAY); } }) .within(Time.minutes(5));后面接上 CEP 的patternStream就能跑。不过 Pattern API 写起来比较啰嗦如果你们的项目里大量使用 FlinkSQL更推荐在 SQL 里直接建 Pattern。Flink 也提供了 Row 类型输入的 Pattern 算子两种方式可以结合。3.2 循环模式与匹配策略times、greedy 的这些参数别搞混除了简单的事件序列CEP 还支持循环模式比如“连续三次登录失败”。这里有几个关键参数times(n)精确匹配 n 次。times(n).optional()匹配 0 到 n 次。oneOrMore()至少一次。zeroOrMore()零次或多次。.greedy()尽可能多地消费事件。.consecutive()要求循环的事件必须严格连续。举个例子pattern.times(2).consecutive()要求两次失败登录之间不能有其他事件如果不加consecutive()那么两次失败中间即使有成功登录也照样匹配。这个区别在安全场景里非常重要因为连续失败一般意味着撞库而中间插入成功登录可能只是用户手误。关于匹配策略Match Skip Strategy很多教程里讲得少但生产环境非常关键。默认策略是NO_SKIP即同一个事件可以参与多个匹配结果这容易产生大量重叠告警。比如你识别“连续三次失败登录”四次连续失败会产生两个匹配结果每个都包含三次事件。如果业务上希望一个事件只用于一次告警就需要用SKIP_TO_NEXT或者SKIP_PAST_LAST_EVENT。这个选择直接决定了你会不会在告警平台里收到重复信息别等到告警轰炸了才回来调。3.3 超时事件的捕获匹配上了但时间不满足了怎么办CEP 还有个容易忽略的点超时事件。within(5 minutes)限定了匹配的时间窗口很快会有一种情况发生模式的前半段已经匹配但后半段迟迟不来直到窗口结束都凑不齐。默认情况下这种超时事件不会被输出你只能拿到“已经完整匹配”的结果。但在很多业务里“超时”本身就是一个很重要的事件。比如你监控的是“下单后 5 分钟未支付”的场景。如果你只用正常匹配输出那么只有支付成功的事件会触发未支付的用户就永远不在输出里出现。这时候就必须用within配合超时侧输出流Side Output来捕获。PatternTimeoutFunction 允许你把“窗口结束但模式未完成”的事件单独输出到一个标签里然后再做后续处理比如发一条催付通知。这段代码里你会看到“outputTag”这个东西OutputTagOrderEvent timeoutTag new OutputTagOrderEvent(timeout) {}; SingleOutputStreamOperatorOrderEvent process CEP.pattern(stream, pattern) .inProcessingTime() .flatSelect( (PatternFlatSelectFunctionOrderEvent, OrderEvent) (map, out) - { out.collect(map.get(start).get(0)); }, PatternTimeoutFunctionOrderEvent, OrderEvent (map, timestamp, out) - { out.collect(map.get(start).get(0)); } );这个侧输出流在 SQL 里的对应物是TIMEOUT检索语句。FlinkSQL 的模式识别里有WITHIN子句时会有一个_Timeout_标号表示超时结果你可以用SELECT * FROM pattern MATCH_RECOGNIZE把超时事件单独捞出来。这块语法相对新但很实用能直接在 SQL 层解决“下单未支付”“物流超时未签收”一类的问题不用再写一堆 Java 代码。4. 两者结合一个订单超时未支付告警的完整实操4.1 需求拆解把业务语言翻译成 Flink 术语为了说明 SQL 和 CEP 在实际项目中怎么协作我拆一个真实场景某模拟订单系统需求是“如果用户下单后 15 分钟内没有支付就发送一条告警”。另外补充一个条件如果该用户在 15 分钟内取消了订单就不需要告警因为业务上已经关闭了这笔订单。这个需求表面上看很简单细想之后有两个难点。第一你需要把一个订单从创建到支付或取消的整个生命周期串起来而且事件可能乱序订单可能先创建后支付也可能先支付后创建极端情况下支付回调先到达订单创建消息晚到。第二告警需要基于“没有支付”这个负向结果如果只用 SQL 做关联你会发现很难表达“某个时间内不存在支付事件”。最终方案是先 SQL 清洗和去重把订单创建事件、支付事件、取消事件统一到一个 Kafka topic 里转化成统一的 Flink 内部 Row然后用 CEP 定义一条模式链CREATE 事件开始15 分钟内既没有 PAY 也没有 CANCEL就输出超时告警。这样分工很清晰SQL 负责“洗干净”CEP 负责“识别时序”。4.2 用 FlinkSQL 做预处理统一事件格式和去重我在实际中用了 Source 表加一条简单 SQL 来统一事件类型。源表结构大概是这样的CREATE TABLE raw_event ( order_id STRING, event_type STRING, op_time BIGINT, rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(op_time / 1000)), WATERMARK FOR rowtime AS rowtime - INTERVAL 3 SECOND ) WITH (...);然后我再建一张 upsert 表或者 Kafka Append 表把订单 id 作为主键同时保留最新的事件类型和时间。这一步主要是为了过滤掉重复事件避免 CEP 因为同一个订单的重复 CREATE 事件而误匹配多次。SQL 里可以用SELECT order_id, event_type, MAX(rowtime) ... GROUP BY order_id, event_type简单去重。但要注意简单的 GROUP BY 会把多行合成一行CEP 需要的是每一条事件独立流入所以我在实际里并不对这种结果做“合并最新”的处理而是用 ROW_NUMBER 根据事件时间取每个订单每个事件类型的最新一条还保留原有的事件流顺序。这个环节最容易踩的坑是用了 SQL 之后字段类型全变了。比如 BIGINT 的 op_time 经过TO_TIMESTAMP后是 TIMESTAMP(3) 类型CEP 里比较时间的时候要用event.getTimestamp或者把 SQL 处理后的 TIMESTAMP 转换成 LONG 再比较。类型不一致会导致后续 Pattern 里面查询字段时出现空指针或者类型转换异常。4.3 CEP 模式实现CREATE 开始15 分钟等待事件预处理后的流进入 CEP核心 Pattern 是Pattern .OrderEventbegin(create) .where(event - event.eventType CREATE) .followedBy(pay_or_cancel) .where(event - event.eventType PAY || event.eventType CANCEL) .within(Time.minutes(15))注意这里我用的是followedBy因为真实场景中 CREATE 和 PAY 之间可能还夹着一些其他事件如查询库存、修改地址如果用next会导致很多本应正常的订单被判断成超时。然后设置超时侧输出把 15 分钟内没有匹配到 PAY 或 CANCEL 的订单捞出来。这里有一个业务细节within(15 min)匹配到 PAY 的事件才是正常匹配输出如果超时侧输出里的订单说明它在 15 分钟内既没有 PAY也没有 CANCEL就可以发告警。如果你把 CANCEL 当作正常结束事件也放进来那么 15 分钟内的取消订单就不会进告警正好符合需求。从这段实操可以看出SQL 用在 CEP 前面做数据清洗和结构化CEP 用在最后做模式判断二者并不冲突。如果你整个任务只用 CEP不洗数据你会发现会因为重复事件、乱序事件、字段类型不一致导致大量误报如果只用 SQL你很难写出“15 分钟内没有支付也没有取消”这句话因为 SQL 默认只表达“存在”不好表达“不存在且不超时”。5. 常见问题与排查技巧实录5.1 状态越来越大CEP 任务内存持续上涨CEP 任务需要缓存匹配中的事件状态比如等待 PAY 的 CREATE 事件这在业务高峰期会造成状态膨胀。首先要检查你的 Pattern 里有没有循环模式、有没有oneOrMore()这类模式状态增长非常快。其次是确认within时间是不是合理如果一个模式允许 1 小时的匹配窗口那么这一小时内所有未完成事件都会积压在状态里。如果你发现状态一直在涨一个常见的原因是事件时间字段没有正确设置水位线导致 CEP 的时间推进很慢within迟迟不触发旧事件没法清理。解决思路是检查源表 Watermark 是否正常推进以及下游有没有设置空闲源超时。还可以考虑用 RocksDB 状态后端把状态放到磁盘上不要让所有状态都堆在堆内存里。根据我的经验生产环境里 CEP 任务状态大、GC 频繁几乎都是 Watermark 不推进 状态后端配置不当这两个原因叠加造成的。5.2 SQL 函数和 CEP 模式匹配出来的结果对不上有些同学会遇到这种情况SQL 里统计每个用户的订单数是正常的但 CEP 匹配出的“订单未支付告警”数量却偏多。原因一般在前置处理上。CEP 是严格依赖事件顺序的如果上游 Kafka 是多分区并行读取你得到的事件流顺序在分区层面是有序的但在全局层面是乱的。你需要先在 SQL 里做一次 key 级别重新分区或者直接在 CEP 前按 order_id 做 keyBy否则跨分区的订单事件可能会乱序导致本应匹配的模式被判定为超时。这里有一个经验参考CEP 的 Pattern 里涉及多个事件时务必按业务主键 keyBy例如按 order_id 或者 user_id。多事件之间如果跨越了多个 Kafka 分区顺序很难保证你会在排查时发现“有时候告警有时候不告警”看起来像偶现 bug其实是数据分区搞得鬼。5.3 侧输出流一直为空超时事件查不出来看过很多同学的 CEP 作业明明设置了 PatternTimeoutFunction但侧输出流就是没数据。排查思路有三步。第一步确认within的时间确实小于业务事件间隔如果写成了 15 小时而不是 15 分钟几乎不会超时。第二步确认你是用inEventTime()还是inProcessingTime()Process Time 模式下机器时钟走得很快超时很敏感Event Time 模式下要确认 Watermark 是否真的到了窗口结束时间。第三步确认你查的是侧输出 Tag而不是主输出流。很多人在测试时只看主流忽略侧流结果。这三个步骤里第二步最隐蔽。你明明设置within(15 minutes)但事件时间水位线就一直没有推进那么所有事件都不会被判定为“超时”因为超时的触发条件是“当前事件时间超过了最后进入模式的事件时间 15 分钟”。如果你没往 CEP 里发送任何水位线推进消息它就永远不触发。这时候你需要观察 JobManager 日志里的 Watermark 数值并检查上游任务是不是有分区空闲。实操总结中的几点个人体会写到这里我想分享三个从实践中悟到的东西。第一FlinkSQL 和 CEP 不是互斥选项。我一直把它们看成是一条流水线上的两道工序SQL 解决“数据怎么洗干净、怎么算快一点”CEP 解决“事件形状怎么识别”。很多复杂需求单独用一种很别扭组合使用反而特别顺。第二SQL 里的时间字段类型要提前设计好CEP 里对时间戳的要求往往是毫秒时间戳。你在 SQL 里转换成 TIMESTAMPCEP 端还要再转回 LONG这中间转换函数写错排查会花很多时间。早早在源表把这些字段命名统一后面整个链路都轻松。第三不要忽视 Watermark 在跨算子之间的传递。SQL 里设置的 Watermark 到了 CEP 算子那边需要通过流记录和时间戳字段继续传递。你如果发现 CEP 超时判断跟预期差很多优先检查 Watermark 有没有真正从上游推进到下游而不是一上来就怀疑 Pattern 写错。另外补充一点关于调试的小技巧CEP 作业在测试环境里难以通过“只喂几条数据”来判断准确性我习惯在本地使用一个模拟事件发生器按照乱序、重复、缺失、超时四个维度随机生成数据再配合 SQL 预处理这样能最大程度还原线上行为。你如果接手一个实时任务多花半小时做这种模拟测试往往能省掉后面几天的排障时间。Flink 的学习是螺旋式上升的Day 79 这个阶段能把 SQL 和 CEP 结合起来用才算是真正跨过了实时计算的“工具拼装期”进入“业务建模期”。希望这篇内容对你的实时任务设计有帮助。
