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

问题导读


1.CEP系统的构建思路是怎样的？
2.主流开源CEP技术框架有哪些？
3.闲鱼CEP系统如何设计实现？



背景
复杂事件处理（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. 事件标准化输入：

[mw_shl_code=text,true]输入样例：        
event_code:  S_SEND_MSG        
event_time:  2019-09-24 10:15:23.474        
extra_info:  {"to_user":"ccc","ARG3":"0",        
              "SDKTYPE":"mini","UTPVID_T":"156*****104","ARG1":"xchat",        
              "chat_session_type":"1","APPKEY":"***","EVENTID":"***",        
              "PAGE":"UT","_priority":"4","chat_session_id":"*****",        
              "content_type":"1",        
              "text":"******"}        
user_id:  abc[/mw_shl_code]

2. 事件的标准化输出：默认输出到消息服务中。消息体是由输出字段组成的json kv结构
[mw_shl_code=text,true]输入样例：        
{"user_id":"abc","event_time":"2019-09-24 10:15:23"}[/mw_shl_code]

运行架构

我们定义了一种简单的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语法类似，第二原则是要足够的简单明了。结合闲鱼的使用场景，我们定义的主要语言要素如下：
[mw_shl_code=text,true]--语言定义        
RULENAME <规则名称>         --定义规则名称        
<变量名:事件类型>           --定义事件变量        
EVENT    <事件模式序列>     --定义事件匹配模式        
[WHERE]  <条件表达式>       --定义事件过滤条件        
[REPEAT] <频次表达式> [SAME] <指定REPEAT重复计数字段> [HAVING] <条件表达式> --定义频次表达        
[WITHIN] <时间窗口>        --定义匹配事件窗口        
RETUEN   <输出字段>        --定义输出事件的字段        
支持场常见内置函数：SUM、COUNT、MAX、MIN、DISTINCT[/mw_shl_code]

整个语言的定义比较简洁，熟悉SQL的同学非常容易理解其所要表达的含义。其特点如下：
1. 简化了复杂事件处理的表达。
2. 标准化的事件输入与输出。
3. 语法与关键词含义类SQL。
4. 统一表达方便云端共用。
5. 只表达事件的处理、触发计算，包含基本的过滤、聚合、模式匹配。
6. 支持常见SQL聚合函数。
7. 非通用编程语言，不覆盖所有CEP场景。
编程示例：
[mw_shl_code=text,true]--用户，在10分钟内，发送了相同条消息，给了100不同的用户。        
RULENAME: "示例"        
e1: S_SEND_MSG        
EVENT  e1        
REPEAT 100+ SAME e1.user_id HAVING DISTINCT(e1.extra_info[received_user_id]) > 100        
WITHIN 10 MINUTE        
RETURN   userId=e1.user_id[/mw_shl_code]

DSL语言实现简介
该DSL语言在端上的实现还在开发中，本文只介绍其在云端的实现方式。由于我们选择的计算引擎是Blink，自然该语言需要能够运行在Blink平台上。Blink底层已经支持CEP计算，其JAVA API说明可参考：https://ci.apache.org/projects/f ... 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。

闲鱼团队是Flutter+Dart FaaS前后端一体化新技术的行业领军者，就是现在！客户端／服务端java／架构／前端／质量工程师面向社会招聘，base杭州阿里巴巴西溪园区，一起做有创想空间的社区产品、做深度顶级的开源项目，一起拓展技术边界成就极致！

最新经典文章，欢迎关注公众号

来源：CSDN

作者：闲鱼技术

原文：《闲鱼如何打造高效CEP系统及DSL编程语言》

https://blog.csdn.net/weixin_38912070/article/details/102597904?utm_source=app

感谢分享