端智能系列文章|端侧复杂事件实时处理框架

背景

现在移动网络越来越发达，移动生活越来越丰富，在用户手机上可能同时存在数百种APP，这注定了用户使用某一款APP的时间也将逐渐缩短。如果用户在APP内仅浏览了几分钟甚至几十秒，那我们将很难为用户提供更有价值的服务与信息，大部分应用的做法是将最最热销的产品或最最火爆的活动放在应用的闪屏或是首焦上，对于闲鱼这样具有丰富业务生态的应用来说，显然还不够。
那如何在用户驻留的短暂时间内为用户提供更有价值的信息呢，闲鱼基于用户的访问路径，为用户提供了更加丰富的优质信息与服务。例如用户发布了商品，我们会推荐当前正在进行的包邮活动，用户没买到想要的商品，我们会及时给用户推荐同款。
为了实现这样的信息提供模式，闲鱼打造了一套流量调控系统，本篇将介绍该系统端侧部分的实时处理框架。

实时处理框架是什么

闲鱼实时流量调控系统的端侧版本是一个合作项目，我们称之为BehaviR，BehaviR框架在纵向流程上，涵盖了用户行为的实时处理，高质量的数据供给，强实时的数据传输，实时计算，决策与用户触达，在广度上能支撑用户在应用内的全场景，同时对典型场景会有定制的支持。而本篇介绍的实时处理框架即是其中的一部分，负责实时计算的闭环，让我们看下BehaviR全貌：

为了完成实时计算的闭环，实时处理框架需要具备的能力：

在实时处理框架中，负责规则计算的部分即是端侧计算引擎，同云侧计算引擎一样，端侧计算引擎的目的是也将获取到的无终结状态的数据进行有状态的实时流式计算。但由于计算发生在端侧，在一个输入源相对有限、运算资源相对有限、有版本限制的运算环境下，要完成上述目的，它需要解决以下几个问题：

• 如何在端侧保证无终结状态的数据持续供给

• 如何在端侧进行有状态的实时计算

• 如何将实时计算结果进行有效输出

数据

在端侧，数据的供给流程如图：

数据的供给能力由BehaviR的数据模块完成，业务方在APP内集成BehaviR，然后将页面操作数据、网络请求数据、存储操作数据等有序灌给BehaviR，BehaviR会根据计算配置将待计算数据按需进行持久化和供给。例如端上指定如下配置：

{
"actionTypeIn":[
"leave"
],
"sceneIn":[
"https://market.m.taobao.com/app/idleFish-F2e/idlefish-renting/home",
"https://market.wapa.taobao.com/app/idleFish-F2e/idlefish-renting/home"
],
"taskArray":[
{
"taskType":"py_backtrace",
"pythonName":"test_cep_rent_rule2",
"filter":[
{
"alias":"e1",
"actionType":"leave"
},
{
"alias":"e2",
"actionType":"pv"
}
]
}
]
}

我们会在"actionType"内配置触发器，即BehaviR将会在"actionType"为"leave"的事件发生时，将拥有的数据提供给计算引擎。在计算发生前，计算所需要的数据类型是可预期的，因此BehaviR会根据"filter"内的配置，将数据过滤后传输给计算引擎。

在APP中运行计算任务前，每一个计算任务都将拥有一份这样属于自己的配置，来指定需要的计算数据类型。在BehaviR中根据端侧所有计算任务的数据需求，可以择优进行持久化来尽可能少的占用磁盘空间和内存空间。

计算

云侧实时计算能力已经成为了业界先驱，Flink、Siddhi、Spark等如数家珍，而端侧并没有计算框架先例，那我们在端侧需要做的，就是深入了解云侧引擎，然后针对端侧环境进行轻量化和定制化。结合我们已承接业务的经验，将计算框架从功能层面划分如下，并与云侧进行比较：

实时计算的窗口，可以对数据进行一定的预处理，例如按时间间隔每N秒处理一次，按数量每N个事件处理一次等。而在端上由于每个行为都有典型的特征，而大多数场景都对这些特征有着明确的预期，因此端上以Trigger的方式进行事件的处理，在Trigger上会描述我们需要开始计算的事件具有的属性，当事件属性与Trigger的属性匹配时，即开始计算。
（不确定有限）状态机和共享缓冲区是计算处理的核心，其中共享缓冲区是对状态计算中状态与事件对应关系的存储优化，我们在初版上也实现了其必要的组成部分。
在并发计划和事件时序控制上，端上暂时通过单端事件的有序录入，来保障绝大多数事件的时序，而并发的优化并不是初版的瓶颈所在，于是我们直接放弃了并发控制，在单线程上进行计算。
简而言之，我们去掉了多流处理的设计，在容错机制上弱化处理，但保留了核心的计算能力，同时将实时流处理上的概念进行融合，更加贴近我们的应用场景。关于计算部分的具体实现，后续将提供更详细的介绍。

实现方案选型

当框架的概念与功能确定后，我们需要考虑使用什么样的方式去实现。于是将预备的实现方案从动态性、执行效率、开发效率等方面进行了比较：

由于在整个端侧处理框架研发上，我们是首次尝试落地，期望能快速验证可行性，因此项目上线及运行过程会具备很高的不确定性及稳定性风险。综上几个因素，恰逢集团能拥有高效、稳定的Python运行时框架Walle，因而我们选择在Python运行时框架下Walle，使用Python进行研发。

描述及编译

在开始运算前，我们会有一套基于Python的应用编程接口来描述计算逻辑，当使用该接口编写完成后，会在运行时编译并构建计算图实例，然后进行计算。
该编程接口的表达以“访问详情然后离开”为例，我们可以描述为：

Pattern('e1') \
.where(KVCondition('actionType', 'pv')) \
.and(KVCondition('scene', 'item_detail')) \
.followby('e2') \
.where(KVCondition('actionType', 'leave')) \
.and(KVCondition('scene', 'item_detail'))

为了云与端的体验一致性，我们也将支持从调控系统的标准DSL（该DSL已成为Blink支持的标准）到Python描述的转换，上述代码用DSL来描述，即：

EVENT: e1->e2
WHERE e1.extra_info.actionType = 'pv' 
AND e1.extra_info.scene = 'item_detail' 
AND e2.extra_info.actionType = 'leave'
AND e2.extra_info.scene = 'item_detail' 

输出

当计算完成后，我们会将计算结果进行输出，在端侧的输出主要分为三个方面，一是业务权益触达，二是算法模型输入，三是另一次计算输入。
由于在业务权益触达形式上，不同的应用有着自己的展现形式和实现方式，因此我们仅保留公用的通讯协议来将计算结果进行格式化输出。在闲鱼端内，我们会介入协议的响应方，并接入闲鱼的决策分发模块进行统一管控。决策分发模块将对单次策略的触达效果进行管控，同时对同一用户的所有策略进行调控。在闲鱼应用内已经注入了丰富的触达形式供决策模块选择。
端侧的算法模型运行时，经常需要特定的行为作为模型运行的触发器，那么当计算结果作为算法模型输入时，由于在集团内端侧算法模型有统一的运行平台，因而我们需要将计算输出与统一的模型运行框架对接，支持特定的计算结果自动唤醒对应模型的运行，然后将计算结果经过有选择性的筛选后输出给算法模型。
如果需要通过多个计算策略串联计算后产出结果，那我们会提前约定好计算串联前的数据规范，当另一次计算处理需要以当前计算结果作为输入时，我们会在本次计算结果模拟为一次约定的特殊类型的行为数据并流入数据模块，然后在另一次计算配置上添加该类型数据的监听，即可顺利执行。

流程回顾

当我们了解了端侧复杂事件实时处理框架的各个部分后，可以用下图再回顾下整个运行过程：

我们会在应用启动的时候去同步本次启动需要进行的计算任务配置，同时数据模块会开始实时处理数据，当处理的数据与任务配置相匹配时，即开始给计算框架供给数据，然后进行实时计算。计算完成后，会流转决策分发模块，由决策模块决定计算结果的输出方向，如果决策为触达用户，则会采用相关的触达配置与形式进行触达。

效果及展望

我们目前在闲鱼APP线上稳定运行了该实时处理框架的首个版本，云侧需要处理5秒左右，而现在全部流程可在毫秒级内完成，对服务器资源零消耗。

在计算细节上，我们仍然有很多需要打磨的地方，例如端上进程退出带来的计算终断问题， 少量事件乱序问题，需要能承接更多复杂场景的聚合计算等。由此，我们将为了更高效的计算和更稳定的服务而努力。更加详情的计算细节介绍，尽请期待！