写点什么

闲鱼端侧如何实现实时 CEP 引擎

  • 2020-02-16
  • 本文字数:5487 字

    阅读完需:约 18 分钟

闲鱼端侧如何实现实时CEP引擎

闲鱼端侧如何实现实时 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 编写如下所示:


.followed_by('middle').where(SimpleCondition())\.next_('end').where(SimpleCondition())
复制代码


这个代码里面声明了 3 个 Pattern,依次命名为 startmiddleend。Pattern 里面保存了指向前面节点的引用 previous,整个 Pattern 链表构建完如下图所示:



最终拿到的是 end节点的一个引用 Ref,Pattern 中会有一个变量指向前一个节点,这样就可以得到一个 Pattern 的反向链表。


Pattern 的对外接口定义如下:


# 静态方法,用来生成起始的pattern@staticmethoddef begin(self, name):pass# 标记紧接着的事件def followed_by(self, name):pass# 标记不需要紧跟的事件def not_followed_by(self, name):pass# 标记紧跟的事件def next_(self, name):pass# 标记事件循环次数def times(self, times):pass# 标记当前事件触发的条件def where(self, condition):pass# 标记当前事件的and条件def and_(self, condition):pass# 标记当前事件的or条件def or_(self, condition):pass# 用于聚合def group_by(self, fields):pass# 用于聚合,渠道特定字段的值def fields(self, key_by_state_name, field):pass# 用于聚合,统计事件具体的数量def count(self, field, condition):pass
复制代码


不同接口会生成不同的消费策略的节点,具体细节可以参考 StateTransition。有了 Pattern 链表,接下来就需要编译器(Compiler)了,它主要是将 Pattern 链表转化成 NFA 图,首先来看下 NFA 的 2 个核心组件:StateStateTransition

State

结构定义如下:



def __init__(self, name, state_type): self.__name = name # 节点的名称,同Pattern的名称 self.__state_type = state_type # 节点的类型:Start/Normal/Stop/Final self.__state_transitions = [] # 到其他节点的边
复制代码


State 一共有 4 种类型:Start/Final/Normal/Stop


生成 NFA 的过程就是将反向解析 Pattern 链表的过程,大概的过程如下:


  1. 创建一个 $end$的结束节点( Final

  2. 再从后往前创建每个 state 节点,作为中间节点( Normal/Stop

  3. 最后创建一个开始节点( Start)State 的名称就是 Pattern 的节点名称,创建完成之后如下图所示。


Transition

State 代表了当前状态机的状态,不同状态之前的切换定义成 StateTransition


结构定义如下:



def __init__(self, source_state, action, target_state, condition): self.__source_state = source_state # 开始的State节点 self.__action = action # 具体action类型:take/ignore/proceed self.__target_state = target_state # 结束的State节点 self.__condition = condition # 节点之间的判断条件
复制代码


边的生成逻辑跟 Pattern 的事件消费策略相关,以下是事件消费策略:


    STRICT = 0# 严格匹配下个    SKIP_TILL_NEXT = 1# 跳过下一个    SKIP_TILL_ANY = 2# 跳过任意一个    NOT_FOLLOW = 3# 非跟随模式    NOT_NEXT = 4# 非紧邻模式
复制代码


不同的消费策略,得到的状态机如下图所示:



  • STRICT: 如果命中了事件了,会进到下个状态

  • SKIP_TILL_NEXT: 如果命中了会进入下一个状态,否则会再当前节点循环,进入 ignore 的边

  • SKIP_TILL_ANY: 不管是否命中条件,都会一直在当前状态循环

  • NOT_FOLLOW: 如果遇到了一个匹配的,就会进入 Stop 状态

  • NOT_NEXT: 如果命中一条,则进入 Stop 状态


在 Pattern 中,不同的接口会创建出不同的消费策略节点,例如 followed_by接口会创建 SKIP_TILL_NEXT的节点。

Times

如果有的规则,要求特定的事件,循环出现几次,那现在就要用到 times 接口。比如浏览 3 次宝贝这个规则,规则就可以写成:


SimpleCondition
复制代码


最终就会得到一个 Times=3的 Pattern,编译器在拿到这个 Pattern 之后,一样先创建一个的 Final 节点,在处理 times 的时候,会创建重复的节点,只不过名称不同,不同的点之间用 take 链接起来,如下图所示:


Python CEP 聚合

Flink 是通过 InputStream 将匹配的事件转移给 CEPOperator,执行聚合操作;但是在客户端的聚合,一次执行就一个事件流,所以可以将聚合简化到一次匹配过程中,因此我们对于 Flink 的聚合操作做了改造,使其更适合端上的场景。


那么聚合的脚本写法如下:


.followed_by('middle').where(SimpleCondition())\.next_('end').where(self.end_filter)\.group_by('group_by').fields('start', 'userId')
复制代码


这里声明了,以 start节点中的 userId作为聚合的节点,我们就会得到如下的 Pattern链表:



在解析 group_by节点的时候,我们需要做个特殊处理,判断如果有聚合节点,我们就需要再 $end$节点和前面节点之间插入一个聚合的节点和哨兵位节点,哨兵位节点命名为 $aggregationStartState$,最终效果如下图所示:



在 NFA 匹配的过程中,当匹配结束,就可以将匹配到的事件流,传到聚合节点,再进一步聚合。$aggregationStartState$节点和 group_by节点之间,是通过 proceed 结合,不需要满足特定条件就可以执行。


具体的实现过程如下,可见与 Flink 不同的是,我们创建了一个特殊的 State节点 AggregationState


def __create_aggregation_state(self, sink_state):# 渠道聚合节点的condition    _aggregation_condition = self.__current_pattern.get_aggregation_condition()
# 创建AggregationState not_next = AggregationState( self.__current_pattern.get_name(),StateType.Normal, _aggregation_condition.get_key_by_state_name(), _aggregation_condition.get_field()) self.__states.append(not_next)
# 获取take的条件 take_condition = self.__get_take_condition(self.__current_pattern) not_next.add_take(sink_state, take_condition)
# 将游标指向上一个节点 self.__following_pattern = self.__current_pattern self.__current_pattern = self.__current_pattern.get_previous()
return not_nex
复制代码

Show me the code

讲了太多原理的东西,接下来看下代码里面如何工作的,先来看下如何来编写一个 CEP 策略。

策略脚本

现在看下如何写一个完整的 python 版本的 cep 规则,以宝贝详情页为例,规则描述如下:


需要匹配用户查看 3 次宝贝详情页


那规则的写法如下:


_pattern = Pattern.begin('e1').where(KVCondition('scene', 'Page_xyItemDetail')).times(3)
# 2. 将需要匹配的事件流_batch_data和待匹配的Pattern# CEP内部会先将pattern转化成NFA,然后再用NFA去匹配事件流_cep = CEP.pattern(_batch_data['eventSeq'], _pattern)
# 用来选择的逻辑def select_function(data):pass
# 3. 匹配完成,通过cep的select接口查询匹配到的结果self.result = _cep.select(select_function)
复制代码


CEP.pattern()函数里面,会先创建 NFA,然后去进行匹配,可见整个匹配策略脚本非常的短小精悍。

生成 NFA

如下代码用来将 Pattern链表转化成 NFA图:


if self.__current_pattern.get_quantifier().get_consuming_strategy() == ConsumingStrategy.NOT_FOLLOW:raiseException('NotFollowedBy is not supported as a last part of a Pattern!')# 校验Pattern的名称,必须唯一self.__check_pattern_name_uniqueness()# 校验Pattern的策略self.__check_pattern_skip_strategy()# 首先创建Final节点sink_state = self.__create_ending_state()# 判定是否有聚合节点if self.__current_pattern.get_aggregation_condition() isnotNone:# 首先创建聚合节点    sink_state = self.__create_aggregation_state(sink_state)# 然后创建聚合几点的起始节点    sink_state = self.__create_aggregation_start_state(sink_state)# 创建状态机中的中间节点,此函数会循环知道Start节点的Patternsink_state = self.__create_middle_states(sink_state)# 最后创建Start节点self.__create_start_state(sink_state)# 根据state列表和window来创建NFAreturn NFA(self.__states, self.__window_time, False)
复制代码

效果

闲鱼已经上了几个策略,整体看来比较稳定,不过还有很多优化的空间。从实测效果来看,端侧从触发策略到执行 Action 用时不会超过 1s,其中还包含了一次网络请求的时间。

性能数据

  • 执行时间



单个脚本,执行时间大概在 100ms 左右。


  • 内存使用



现在内存使用峰值还是比较高,大概在 15M 左右。关于内存过大的问题,目前正在讨论一个方案:Python CEP 可以持久化当前 NFA 的状态,然后再触发策略的时候,只带从未触发过的事件流,避免很多重复计算。之前运行一次脚本要处理 500 个事件,现在可能就缩减到 100 之内,可以极大的减小内存消耗。同时带来另外一个问题,就是执行脚本的都会有一个 IO 操作,耗时会增加。

Flink 与客户端对比

现在对于 Flink 和客户端 Python CEP 做一个简单的对比:



相比 Flink,端侧 CEP 还是有它的优势,在端侧可以直接利用客户端的埋点信息进行计算,运行时长缩减了 80%,而且也支持动态发布。Python 脚本支持 2 端通投,在保证 2 端埋点一致的前提下,也极大的减少了维护成本。

未来

现在端计算还存在很多待优化的地方:


  1. 端计算是用 Python 实现,无法做到像 Flink 的状态机常驻内存,每次都要重新创建匹配,带来了额外的消耗

  2. 在事件流的清洗上面,现在是通过回朔拿到之前的事件流,存在大量的重复计算,后续可以借鉴 Flink 的 Window 机制来进行优化。

  3. 目前编译器暂时还不支持 Group Pattern,后续还要对其进行扩展。

  4. Python 脚本现在还是需要手动编写,后续还可以考虑通过 DSL 来自动生成。


整体看来,Python 脚本执行策略还是有一定的性能损耗,不管是在创建 NFA 或者是匹配过程,后续可以考虑将匹配引擎用 C++实现,然后真正做到常驻内存,从而做到高效的执行效率。后期做到 NFA 持久化之后,C++也可以复用整套持久化协议,从而优化整个引擎的执行效率。除此之外,策略在执行的过程中,还可以考虑用 TensorFlowLite 优化参数策略参数,从而真正做到千人前面的策略。


本文转载自公众号闲鱼技术(ID:XYtech_Alibaba)。


原文链接


https://mp.weixin.qq.com/s/_jPQvX3aiKp4ScnqgHgyow


2020-02-16 10:002961

评论

发布
暂无评论
发现更多内容

【函数式编程实战】(八) 如何将你的代码重构

小明Java问道之路

设计模式 函数式编程 7月月更 签约计划第三季 Lambda表达式

为啥国内大厂都把云计算当成香饽饽,这个万亿市场你真的了解吗

雨果

DaaS数据即服务

在灯塔工厂点亮5G,宁德时代抢先探路中国智造

脑极体

Plato Farm有望通过Elephant Swap,进一步向外拓展生态

鳄鱼视界

Mybatis中LRU缓存实现

Geek漫游指南

mybatis LRU mybatis源码

2200字详细讲解计算机网络的“流量管理员”:QoS 服务质量

wljslmz

流量控制 QoS 网络技术 7月月更

Plato Farm通过LaaS协议Elephant Swap,为社区用户带来全新体验

股市老人

C# 中的转译字符'/b'

陈言必行

7月月更

模块8(消息队列存储消息数据的mysql表格)

Geek_701557

提前批到底影不影响正式批?

KEY.L

7月月更

次轮Okaleido Tiger即将登录Binance NFT,引发社区热议

鳄鱼视界

【函数式编程实战】(九) Optional实战大全

小明Java问道之路

java8 Optional Stream API 7月月更 签约计划第三季

音乐体验天花板!14个网易云音乐的情感化设计细节

张姣发

创新 交互设计 网易云音乐

关于 StatefulWidget,你不得不知道的原理和要点!

岛上码农

flutter ios 前端 安卓开发 签约计划第三季

平成千字文(へいせいせんじもん)    (平成12年9月10日 石渡 明 作)  宇宙広遠 銀河永久 日月運行 不乱無休 地球公転 季節変移 黄道星座 太陽年周 故郷群島 南熱北冷 海洋温暖 気候順良 青空飛雲 諸野深緑 湖泉静息 谷川清流 春桜一面 新芽

贾献华

7月月更

缓存一致性与内存屏障

蝉沐风

volatile 内存屏障 缓存一致性 MESI

Bootstrap的导航元素和Well详解【前端Bootstrap框架】

恒山其若陋兮

7月月更

数据资产管理的概念

奔向架构师

数据资产 7月月更

客户案例 | 关注老年用户体验,银行APP适老化改造要避虚就实

易观分析

用户体验 银行app 老龄化

C# 窗体应用常用基础控件讲解(适合萌新)

IC00

C# 7月月更

小程序怎样助力智能家居生态新模式

Geek_99967b

物联网

【函数式编程实战】(七) Collection在Java8和9中的增强

小明Java问道之路

Lambda collection Stream API 7月月更 签约计划第三季

一骑入秦川——浅聊Beego AutoRouter是如何工作

Regan Yue

Go 源码刨析 Go web 签约计划第三季 Beego

Prometheus 运维工具 Promtool (三)Debug 功能

耳东@Erdong

Prometheus 7月月更 Promtool

传统车企数字化转型如何打通最后一公里?

雨果

车联网 DaaS数据即服务

次轮Okaleido Tiger即将登录Binance NFT,引发社区热议

股市老人

Okaleido Tiger 7.27日登录Binance NFT,首轮已获不俗成绩

BlockChain先知

React 学习记录2📝

程序员海军

React 7月月更

聊聊自动化测试的度量指标

老张

自动化测试 质量度量

闲鱼端侧如何实现实时CEP引擎_容器_景松_InfoQ精选文章