闲鱼端侧如何实现实时 CEP 引擎
背景
用户来闲鱼,主要是为了获得自己关心的内容。随着闲鱼的体量越来越大,内容也变得越来越丰富。闲鱼基于用户画像,可以将用户关心的内容推送给用户。具体在哪些场景下才需要触发推送?我们定义了很多触发规则,包括停留时长、点击路径等。起初我们把触发规则的逻辑放在服务端(Blink)运行。但实践下来发现 Blink 存在诸多限制:
服务端要对客户端埋点进行数据清洗,考虑到闲鱼的 DAU 已经突破 2000w,这个量是非常庞大的,非常消耗服务端资源;
Blink 的策略是实时执行的,同样因为资源问题,现在只能同时上线十几个策略。
如何解决这些问题呢,我们开始考虑能否将 Blink 的策略跑在客户端!
CEP 模型
Blink,作为是 Flink 的一个分支,最初是阿里巴巴内部创建的,针对 Flink 进行了改进,所以我们这里还是围绕 Flink 讨论。CEP(Complex Event Process)是 Flink 中的一个子库,用来快速检测无尽数据流中的复杂模式。
Flink CEP
Flink 的 CEP 的核心是 NFA(Non-determined Finite Automaton),全称叫不确定的有限状态机。提到 NFA,就不得不提 Jagrati Agrawal 等撰写的关于 NFA 模型的论文《Efficient Pattern Matching over Event Streams》,本篇论文中描述了 NFA 的匹配原理。
上面这张图,就是一个不确定的有限状态机,它由状态(State)还有之间的连线(StateTransition)组成的。
状态(State):状态是根据 flink 脚本里面的代码来决定的,最终会有一个
$end$
的 Final 状态转换(StateTransition):State 的转换条件,包括
take/proceed/ignore
不同的条件,代表的含义不同:
take
: 满足条件,获取当前元素,进入下一状态proceed
:不论是否满足条件,不获取当前元素,直接进入下状态(如 optional)并进行判断是否满足条件。ignore
:不满足条件,忽略,进入下一状态。
我们只要在端上实现这样一个状态机,就可以实现一个 CEP 引擎。
Python CEP
对于客户端来说,首先要解决的问题是如何构建一个 CEP 环境。经过调研,可以复用集团的端智能容器(Walle),作为 Python 容器可以执行 cep 的策略。
在构建 NFA 之前,首先要解决的一个问题是数据来源,手淘信息流团队有一套完整的解决方案 BehaviX/BehaviR,可以对 UT 埋点进行结构化,能很好的结合 Walle 容器来触发策略。有了事件来源,还需要解决的是 Python 脚本如何执行。Walle 平台可以将多个 Python 脚本打包下载并执行,因此,我们可以将 CEP 封装成一个 Python 的库,然后跟策略脚本一起下发。
最终的整体架构设计如下图所示:
本文重点介绍下如何用 Python 来实现一个 CEP 的编译器,这个编译器主要用来将 CEP 的描述语言转换成为 NFA。
编译器原理
在 Flink 中,java 侧会有一套完善的 API 来编写一个策略脚本,《efficient Pattern Matching over Event Streams》论文中还定义了一套完备的 DSL 描述语言,也是会转化成 java 文件去调用这些 API 去完成匹配。那么接下来会重点讨论,flink 是如何将上述 API 转化成 NFA 去匹配,以及 Python CEP 如何实现上述一套完整 API 接口。
Pattern
在 Flink 里面,是通过 Pattern
来构建这个 NFA,首先用它描述这个不确定性状态机。首先是构建一个 Pattern
的一个链表,得到这个链表之后,会将每个 Pattern 映射成为 State
的图,点与点之间会通过 StateTransition
来连接。以下面的 Python 代码为例,看下如何 API 是如何工作的:
例如,需要创建这样一个规则,描述如下:
以 start 事件开始,后续跟随一个 middle 的事件,后面紧跟着一个 end 事件作为结尾
用 Pattern 编写如下所示:
这个代码里面声明了 3 个 Pattern,依次命名为 start
、 middle
、 end
。Pattern 里面保存了指向前面节点的引用 previous
,整个 Pattern 链表构建完如下图所示:
最终拿到的是 end
节点的一个引用 Ref
,Pattern 中会有一个变量指向前一个节点,这样就可以得到一个 Pattern 的反向链表。
Pattern 的对外接口定义如下:
不同接口会生成不同的消费策略的节点,具体细节可以参考 StateTransition
。有了 Pattern 链表,接下来就需要编译器(Compiler)了,它主要是将 Pattern 链表转化成 NFA 图,首先来看下 NFA 的 2 个核心组件:State
和 StateTransition
。
State
结构定义如下:
State 一共有 4 种类型:Start/Final/Normal/Stop
。
生成 NFA 的过程就是将反向解析 Pattern 链表的过程,大概的过程如下:
创建一个
$end$
的结束节点(Final
)再从后往前创建每个 state 节点,作为中间节点(
Normal/Stop
)最后创建一个开始节点(
Start
)State 的名称就是 Pattern 的节点名称,创建完成之后如下图所示。
Transition
State 代表了当前状态机的状态,不同状态之前的切换定义成 StateTransition
。
结构定义如下:
边的生成逻辑跟 Pattern 的事件消费策略相关,以下是事件消费策略:
不同的消费策略,得到的状态机如下图所示:
STRICT
: 如果命中了事件了,会进到下个状态SKIP_TILL_NEXT
: 如果命中了会进入下一个状态,否则会再当前节点循环,进入 ignore 的边SKIP_TILL_ANY
: 不管是否命中条件,都会一直在当前状态循环NOT_FOLLOW
: 如果遇到了一个匹配的,就会进入 Stop 状态NOT_NEXT
: 如果命中一条,则进入 Stop 状态
在 Pattern 中,不同的接口会创建出不同的消费策略节点,例如 followed_by
接口会创建 SKIP_TILL_NEXT
的节点。
Times
如果有的规则,要求特定的事件,循环出现几次,那现在就要用到 times 接口。比如浏览 3 次宝贝这个规则,规则就可以写成:
最终就会得到一个 Times=3
的 Pattern,编译器在拿到这个 Pattern 之后,一样先创建一个的 Final 节点,在处理 times 的时候,会创建重复的节点,只不过名称不同,不同的点之间用 take 链接起来,如下图所示:
Python CEP 聚合
Flink 是通过 InputStream 将匹配的事件转移给 CEPOperator,执行聚合操作;但是在客户端的聚合,一次执行就一个事件流,所以可以将聚合简化到一次匹配过程中,因此我们对于 Flink 的聚合操作做了改造,使其更适合端上的场景。
那么聚合的脚本写法如下:
这里声明了,以 start
节点中的 userId
作为聚合的节点,我们就会得到如下的 Pattern
链表:
在解析 group_by
节点的时候,我们需要做个特殊处理,判断如果有聚合节点,我们就需要再 $end$
节点和前面节点之间插入一个聚合的节点和哨兵位节点,哨兵位节点命名为 $aggregationStartState$
,最终效果如下图所示:
在 NFA 匹配的过程中,当匹配结束,就可以将匹配到的事件流,传到聚合节点,再进一步聚合。$aggregationStartState$
节点和 group_by
节点之间,是通过 proceed 结合,不需要满足特定条件就可以执行。
具体的实现过程如下,可见与 Flink 不同的是,我们创建了一个特殊的 State
节点 AggregationState
:
Show me the code
讲了太多原理的东西,接下来看下代码里面如何工作的,先来看下如何来编写一个 CEP 策略。
策略脚本
现在看下如何写一个完整的 python 版本的 cep 规则,以宝贝详情页为例,规则描述如下:
需要匹配用户查看 3 次宝贝详情页
那规则的写法如下:
在 CEP.pattern()
函数里面,会先创建 NFA
,然后去进行匹配,可见整个匹配策略脚本非常的短小精悍。
生成 NFA
如下代码用来将 Pattern
链表转化成 NFA
图:
效果
闲鱼已经上了几个策略,整体看来比较稳定,不过还有很多优化的空间。从实测效果来看,端侧从触发策略到执行 Action 用时不会超过 1s,其中还包含了一次网络请求的时间。
性能数据
执行时间
单个脚本,执行时间大概在 100ms 左右。
内存使用
现在内存使用峰值还是比较高,大概在 15M 左右。关于内存过大的问题,目前正在讨论一个方案:Python CEP 可以持久化当前 NFA 的状态,然后再触发策略的时候,只带从未触发过的事件流,避免很多重复计算。之前运行一次脚本要处理 500 个事件,现在可能就缩减到 100 之内,可以极大的减小内存消耗。同时带来另外一个问题,就是执行脚本的都会有一个 IO 操作,耗时会增加。
Flink 与客户端对比
现在对于 Flink 和客户端 Python CEP 做一个简单的对比:
相比 Flink,端侧 CEP 还是有它的优势,在端侧可以直接利用客户端的埋点信息进行计算,运行时长缩减了 80%,而且也支持动态发布。Python 脚本支持 2 端通投,在保证 2 端埋点一致的前提下,也极大的减少了维护成本。
未来
现在端计算还存在很多待优化的地方:
端计算是用 Python 实现,无法做到像 Flink 的状态机常驻内存,每次都要重新创建匹配,带来了额外的消耗
在事件流的清洗上面,现在是通过回朔拿到之前的事件流,存在大量的重复计算,后续可以借鉴 Flink 的 Window 机制来进行优化。
目前编译器暂时还不支持 Group Pattern,后续还要对其进行扩展。
Python 脚本现在还是需要手动编写,后续还可以考虑通过 DSL 来自动生成。
整体看来,Python 脚本执行策略还是有一定的性能损耗,不管是在创建 NFA 或者是匹配过程,后续可以考虑将匹配引擎用 C++实现,然后真正做到常驻内存,从而做到高效的执行效率。后期做到 NFA 持久化之后,C++也可以复用整套持久化协议,从而优化整个引擎的执行效率。除此之外,策略在执行的过程中,还可以考虑用 TensorFlowLite 优化参数策略参数,从而真正做到千人前面的策略。
本文转载自公众号闲鱼技术(ID:XYtech_Alibaba)。
原文链接:
https://mp.weixin.qq.com/s/_jPQvX3aiKp4ScnqgHgyow
评论