闲鱼如何打造高效CEP系统及DSL编程语言

背景

复杂事件处理（Complex Event Processing，以下简称CEP）是一种分析事件流的技术，事件可以是事物有意义的状态变化也可以是事物之间的动作。主要用于分析事件之间的关联关系，且这种关联关系在时间上是多种多样的。

在闲鱼内部，需要处理复杂事件的场景逐渐增多。比如：
1. 安全治理领域：
用户N分钟内发送图片或文字消息给M个人，标记该用户。
卖家发布商品，买家快速拍下商品。 M小时内，同一买家超过K次。标记该买家。
2. 营销领域：
12小时内，买家查看商品详情，卖家降价商品。给买家发送降价push。
这些需求常常多变的，而且需要快速上线验证，如何提升开发效率同时满足事件计算的及时性、可靠性、稳定性成为摆在闲鱼面前的一大问题。

CEP系统的构建思路

CEP技术是一项历史比较久的技术，早年就被应用于很多计算机领域-比如应用于无线射频识别（Radio Frequency Identification，RFID）领域的的事件监控预警。
对闲鱼来应用场景来说，总结核心诉求如下：

1. 需要满足常见的计算需求：需要考虑事件序列、窗口聚合、事件过滤、模式匹配计算场景。

2. 开发效率：需求常常是多变的，要求可以低成本开发快速上线。

3. 性能：闲鱼存在海量的用户行为数据以及各种预定义事件，要求千万级/秒的事件处理吞吐量，秒级延迟。

4. 容错性：数据不可丢，计算不可以出错，出现错误后系统可自动恢复。

5. 云端联动：有一些简单计算场景可以直接在端上实现掉，提升响应性能。只有复杂场景才需要在服务端计算实现。

6. 能够解决大部分计算场景就可以了，不需要为了一些非常少见的场景去设计特别复杂的机制。

主流开源CEP技术框架

目前主流的CEP技术花样繁多，各成体系，我们调研了业内主流开源技术方案做了一些比较：

闲鱼CEP系统的思路

调研众多技术方案和参考相关论文后，闲鱼决定采用如下的方式构建CEP系统：

1. 标准化事件的输入与输出，这样可以大幅度降低系统的复杂性。输出事件写入消息服务中供使用方订阅。

2. 为了简化开发工作，同时统一云和端的匹配规则表达，参考业内论文，我们决定自定义一种简单的DSL语言来描述规则。该语言应该类似SQL，由有限的关键字组成，让普通开发者一看就懂，同时有足够的表达能力覆盖大部分闲鱼CEP计算场景。

3. 关于计算引擎的选择：考虑到系统需要处理海量的用户行为数据，对性能、实时性、容错性的严苛要求以及人员技术熟悉情况。选择阿里内部较为成熟Blink（开源版本叫Flink）实时计算框架作为底层计算引擎。

设计实现

输入输出

整个系统本质上是一个数据处理系统，自然首先需要定义的事情就是：输入与输出。

我们抽象了闲鱼的常见的用户行为和动作，作为基本事件，并把这些基本事件标准化。这些基本事件可以看作是我们CEP计算的输入，由基本事件匹配生成的结果可以看作是输出事件。
1. 事件标准化输入：

1. 输入样例：
   

2. event_code: S_SEND_MSG
   

3. event_time: 2019-09-24 10:15:23.474
   

4. extra_info: {"to_user":"ccc","ARG3":"0",
   

5. "SDKTYPE":"mini","UTPVID_T":"156*****104","ARG1":"xchat",
   

6. "chat_session_type":"1","APPKEY":"***","EVENTID":"***",
   

7. "PAGE":"UT","_priority":"4","chat_session_id":"*****",
   

8. "content_type":"1",
   

9. "text":"******"}
   

10. user_id: abc
    

2. 事件的标准化输出：
默认输出到消息服务中。
消息体是由输出字段组成的json kv结构

1. 输入样例：
   

2. {"user_id":"abc","event_time":"2019-09-24 10:15:23"}
   

运行架构

我们定义了一种简单的DSL语言来表达事件处理规则，有了DSL，用户可以像使用普通的数据库一样编写DSL，然后提交给系统自动运行该DSL语言，运行该DSL的系统持续的监听输入数据做实时匹配，并将结果作为输出事件输出。
Interactive Service：系统对外提交的交互服务。
Strean Source：标准化的事件来源，使用阿里云sls存储。
EPL Parser：负责DSL的语法扫描与代码翻译，将用户编写的DSL解析生成Blink CEP Pattern代码或者Blink SQL。
Job Manager：主要负责将生成的EPL Parser生成的代码或SQL部署到Blink的任务中，包括生成任务与执行计划、提交任务、停止任务、启动任务等等。
Sink输入：事件命中计算规则后生成json格式的输出事件，存储到消息服务MetaQ中。

闲鱼DSL语言

我们设计DSL语言的第一个原则就是尽量和SQL语法类似，第二原则是要足够的简单明了。结合闲鱼的使用场景，我们定义的主要语言要素如下：

1. --语言定义
   

2. RULENAME <规则名称> --定义规则名称
   

3. <变量名:事件类型> --定义事件变量
   

4. EVENT <事件模式序列> --定义事件匹配模式
   

5. [WHERE] <条件表达式> --定义事件过滤条件
   

6. [REPEAT] <频次表达式> [SAME] <指定REPEAT重复计数字段> [HAVING] <条件表达式> --定义频次表达
   

7. [WITHIN] <时间窗口> --定义匹配事件窗口
   

8. RETUEN <输出字段> --定义输出事件的字段
   

9. 支持场常见内置函数：SUM、COUNT、MAX、MIN、DISTINCT
   

整个语言的定义比较简洁，熟悉SQL的同学非常容易理解其所要表达的含义。
其特点如下：

1. 简化了复杂事件处理的表达。

2. 标准化的事件输入与输出。

3. 语法与关键词含义类SQL。

4. 统一表达方便云端共用。

5. 只表达事件的处理、触发计算，包含基本的过滤、聚合、模式匹配。

6. 支持常见SQL聚合函数。

7. 非通用编程语言，不覆盖所有CEP场景。

编程示例：

1. --用户，在10分钟内，发送了相同条消息，给了100不同的用户。
   

2. RULENAME: "示例"
   

3. e1: S_SEND_MSG
   

4. EVENT e1
   

5. REPEAT 100+ SAME e1.user_id HAVING DISTINCT(e1.extra_info[received_user_id]) > 100
   

6. WITHIN 10 MINUTE
   

7. RETURN userId=e1.user_id
   

DSL语言实现简介

该DSL语言在端上的实现还在开发中，本文只介绍其在云端的实现方式。
由于我们选择的计算引擎是Blink，自然该语言需要能够运行在Blink平台上。Blink底层已经支持CEP计算，其JAVA API说明可参考：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/libs/cep.html。
在Blink内部是通过NFA（Non-determined Finite Automaton 不确定有限状态机）来实现CEP技术的，其主要理论依据参考的是论文Efficient Pattern Matching over Event Streams中介绍的模型，有兴趣的同学可以参考。

我们知道标准的SQL是可以直接运行在Blink上面的，Blink没有自己构建一个SQL优化解析器，而是很聪明的选择了Apache Calcite来实现SQL校验、SQL解析、抽象语法树的构建以及SQL优化(底层语法分析实际是通过Javacc实现)。Calcite是一个非常流行的开源SQL实现框架，在很多开源项目中都有应用。这样Calcite在Blink SQL架构中处于核心地位。
闲鱼的DSL相当于扩展了Blink的SQL，通过自定义解析器做DSL的解析、校验同时生成抽象语法树，然后根据语法树生成Blink CEP代码。特别地对于一种简单场景：只有对一种类型的事件做过滤、聚合计算的情况；我们将其直接翻译成Blink SQL，因为Blink SQL底层是Stream API比Blink CEP基于NFA的状态机性能要好。
整体架构如下图所示：

实现流程：

1. 利用calcite定制解析器解析DSL语言，生成类SQL语法树。

2. 语法正确性校验。

3. 通过语法正确性校验生成抽象语法树。

4. 判断如果只有一种类型的事件，通过代码替换模板直接翻译成Blink SQL 转7，否则转5。

5. 翻译成CEP Pattern API代码。

6. 添加标准的输入、输出Stream，构建完整的运行拓扑图。

7. 设定任务运行参数。

8. 通过Blink API提交Blink运行。

应用效果

目前该系统已经上线第一个版本，承接了闲鱼的安全策略检查、实时触达用户以及玩法场景下的规则实时匹配。生成的匹配结果通过写入MetaQ供使用方消费。

效率提升：初步验证实现同样的规则匹配功能，同编写JAVA代码相比使用改DSL语言可以大幅提升开发效率，从接受需求到编写DSL上线验证一般30分钟左右即可完成。
性能：DSL生成的计算任务处理10w左右QPS数据，消耗3个cu，平均延迟1s。
高可靠性：依赖于Blink的高可靠特性，任务的运行自然拥有快速的错误恢复机制以及数据乱序处理能力。
测试运行效果如下图所示：

闲鱼的CEP计算还在不断完善中，同时我们在和Blink团队合作共建该DSL语言，成为Blink应用生态的一部分。计划成熟后将逐步应用于阿里内部其他BU。