基于flink与groovy实现全实时动态规则智能营销与风控系统

前言：本文是对视频课程《基于flink与groovy实现亿用户级实时动态规则智能运营系统》的介绍说明；
本项目极具行业实用价值，可为各企业开发人员提供系统设计思路与灵感，而且，它不光可用于智能运营，也可以应用在实时规则推荐，实时广告推送，实时规则封控，实时交通监控等纪委广泛的场景中；
当然，各类培训机构学员也可以通过学习此项目来丰富自己的就业简历，绝对杀手锏级别！在面试中运用得当，可以起到一锤定音立竿见影的绝杀效果！

如果需要学习本项目，可联系博主

1 项目背景

传统的“精准营销平台”由营销人员基于画像标签数据去圈选人群，进行营销推送，存在不足；
这种传统手段，往往无法抓住那些 “转瞬即逝的营销机会”；

• 如：一个促销热爱型用户，正在浏览包包时，及时推送出正在做促销活动的包包信息，是最有效的；
• 如：一个价格敏感型用户，正在犹豫不决反复查看他购物车中的某目标商品时，适时推送优惠券，是最有效的；

这些场景，显然无法通过营销人员通过人工手段去发现，需要用软件系统自动、实时监控用户的行为，并实时做出判断，并进而驱动营销推送系统推送相关营销内容；
总之，在应对这一类的需求时，以前的传统的“精准营销平台”显得力不从心；

因而，擅长用键鼠改变世界的攻城狮们，决定为公司的推广、运营部门开发一个应对上述场景的自动化、智能的、实时、规则动态可变的营销利器 —— 《实时智能运营系统》

2 需求分析

2.1 营销规则举例

• 搜索兴趣词包含“商务休闲”的用户，如果浏览包含“商务休闲”的商品页面，则触发一个优惠券推送消息；
• 活跃度等级为C的用户，且在规则发布前的3天内，发生过 “搜索雀巢” AND “加购” ，当他再次上线时，触发一个app弹窗消息；
• 活跃度等级为C的用户，规则发布前2天内，发生过A事件3次以上，规则上线后发生了B事件，且之后间隔1天再次发生B事件，则触发；
• 规则运行时，检测到用户发生了下单事件，而在随后的15分钟内没有支付，则触发；
• 规则运行时，一个用户如果搜索的关键词符合某正则表达式，则触发；
• 宝妈用户，在2022-08-07开始，做过A行为3+次，做过（B行为，E行为）序列2+次，则触发；

2.2 规则营销需求的宏观抽象

• 营销规则有多样化的灵活的定义（不光参数可以支持自定义，且运算模型也可以自定义）；
• 营销规则通常针对满足某画像的人群，也可没有针对性
• 营销规则通常都有有效期（也有可能是无限期的），要求系统能够支持动态增加、停用运营计划（规则）
• 营销规则根据需求的变化，需要能够方便地进行在线管理，如新增、停用、修改

2.3 营销规则受众条件的逻辑要素抽象分析

• 各类营销规则，都支持对目标人群的圈选
• 圈选的手段之一是：是对用户的画像标签数据进行定义、判断
• 圈选的手段之二是：对用户的行为数据进行自定义即时计算（广义上就是实时用户画像）
• 而行为数据即时计算的时间跨度，通常是有时间窗口约束的，而所涉及的时间跨度可能有：
• 只包含规则上线之前
• 只包含规则上线之后
• 横跨规则上线前到上线后
• 受众圈选行为画像即时条件中，抽象出来的判断要素主要有- 事件类型及属性是否满足规则（如搜索事件且搜索词是否满足正则、是否包含指定关键字等）- 事件序列是否满足规则（是否依次发生过A-C-E序列）- 事件或事件序列的发生次数是否满足（如X事件是否发生了N次）- 事件的属性统计值是否满足（如X事件中的sum(属性amt) > 100 ，事件X的行为总时长>20）- 事件与事件之间的对比关系，如时间间隔（如：A-B事件的间隔>N）、属性值（统计）对比
• 最后，就是上述各类条件的逻辑组合： 与 、或 、 非、 大于、 小于、 等于……- 条件1： A发生次数 >3- 条件2 ： A事件的max(amt) - A事件的min(amt） > 200- 条件3 ： A - B间隔> 5min 最终规则： 条件1 与 （条件2 或 条件3） 思考： A事件中pageid=“p05” 的次数 > A事件中pageid="p10"的次数 ，是否包含在上述要素中？ A事件中，MAX(属性amt) > 2* MIN(属性amt) ， 是否包含在上述要素模型中？ 各次A事件 中 amt 属性的 最大的前3个值 ，的平均值 >10 ？

3 系统概要设计

3.1 各类受众条件运算的实现逻辑初步设计（实时画像计算条件）

3.1.1 画像条件的逻辑分析

这个好说，就是一个查询画像标签库进行判断的动作

3.1.2 行为判断的逻辑分析

1）行为判断：行为事件的类型及属性的判断
判断用户的某个行为事件，是否是规则条件中要求的事件

• 简单的，可能就是

event_id = "X"  
 AND
properties["p1"] =/>/!=/< "v1"

• 稍微复杂的 ，可能就是

 event_id = "X"  
  AND 
 properties["p1"]  包含 "kw1"

• 再复杂点的，可能就是

event_id = "X"  
 AND 
properties["p1"]  + properties["p2"] > 100   
 AND
properties["p3"] 满足 正则表达式  [1,10]*?a

2）行为序列的判断

判断用户的一系列行为事件中，是否满足规则条件所要求的依次发生的序列，比如A B E

行为序列判断，是上述 “行为判断” 的衍生
本质上，就是在 判断事件 的基础上，加上时间先后的序列关系
这个逻辑，可以用 正则匹配进行计算（全窗口运算）； 也可以用滚动聚合的方式计算；

3）事件（序列）发生次数的判断
判断要求的事件或事件序列，发生了多少次
是上述 “行为判断”、“行为序列判断” 的衍生
本质上，就是在 判断事件（或序列）的基础上，进行次数累计

4）事件属性值统计条件运算
也好说，主要是对满足条件的事件，为目标属性值进行累加等运算
比如： 要求 用户所发生的的一系列A事件中，属性amt最大值 - 最小值 >100
暂时无法在飞书文档外展示此内容
比如： 要求 用户所发生的的一系列A事件中，属性amt最大的前3个，的平均值 > 200

5）事件与事件对比关系的判断
事件对比关系的判断，也是上述“行为判断”的衍生
如事件间隔时长的判断，本质上，也是在 判断事件的基础上，进行时间差的运算

比如，要求， A事件 后面发生 B事件 ，且 时间间隔 要 > 15分钟

结论：不管用户的规则如何变化，如何灵活多样，只要设计好合理的运算逻辑+合理的中间聚合结果数据结构，我们都可以用持续、滚动运算的方式来实现！

3.2 受众条件运算实现逻辑深入设计
3.2.1 静态画像条件的逻辑实现
1）画像条件判断的实现难点
初略考虑，应该就是查询用户画像标签库即可；
站在系统的角度来考虑时： 什么时机去查询？ 去哪里查？

比如： 查询时机： 当一个人的某个行为触发了某个规则，此时去查询该用户的画像标签是否满足规则的人群画像条件 查询的目标：
用户画像标签数据库

如果画像数据库在hbase中，可以用如下伪代码来快速判断

## 如果条件是： 活跃等级 = C   AND     首单时间< "2022-06-01"
get = new Get(guid)
get.addColumn("活跃等级")
get.addColumn("首单时间")

result = table.get(Get)

activeLevel = result.getValue("活跃等级")
firstOrderDate = result.getValue("首单时间")

if(activeLevel = 'C'  &&  firstOrderDate< "2022-06-01") 
return true;

## 如果条件是： 兴趣词 ：包含 "速溶咖啡"
## 这种条件，用hbase的支撑就不太好处理了
// favroitWords : [海底捞, 咖啡伴侣, 小罐速溶咖啡，乔丹 ，篮球 ]
favroitWords = result.getValue("兴趣词")  

但，要考虑到系统性能的话，就没那么简单：

• 如果，我们在流式语境中，对每个用户的每次满足某规则的 “触发条件” 事件，根据规则需求，都需要去请求、查询一次用户画像库，可以想象得出这种查询的超高并发度和请求频率；
• 从而，对画像标签数据的存储backend产生的压力可想而知；
• 就算使用HBASE或者elastic search，能勉强支撑，应付这种高强度的查询请求，对于hbase或elastic search的集群规模要求，可能会超出公司的成本预算；

2）方案概要设计
根据上文的分析，最好避免使用上述的每次规则触发都去查询外部存储系统的方案

画像条件判断的环境特点

• 我们的画像系统，是一个离线画像系统，因而，画像系统中的数据并不是时刻变更的
• 因而，对于一个用户，是否满足某规则的画像条件，完全可以在规则上线前即可确定

方案要点设计
针对上面的特点分析，我们可以采用如下方案来解决规则画像条件判断的需求

1. 在发布规则的时候，根据规则的画像条件，先从画像库中条件查询，圈选满足画像条件的人群
2. 将圈选出的人群的guid，转换成bitmap
3. 将（规则id，对应的画像人群bitmap），注入到我们的实时流式规则计算引擎（flink程序）
4. 实时流式规则计算引擎，则可以在内部持有每个规则的人群bitmap，进而在需要判断的时候，直接在内部持有的规则人群bitmap中判断即可
5. 另外，当外部画像库进行了离线批量更新时，将人群圈选及接入引擎的流程，也进行一次，以更新规则引擎中持有的人群bitmap

3.2.2 即时画像条件分析判断的难点

1）难点概述
即时画像条件的计算在什么时候进行，在哪里进行

• 行为条件的判断计算，是需要依赖用户的持续不断实时输入的行为明细数据的
• 而且，行为条件的计算、判断，可能涉及3种类型的时间跨度

只涉及规则上线之前
只涉及规则上线之后
横跨规则上线的前后

综上，我们的规则引擎在流式计算的语境下，该去哪里查询用户的明细数据，就需要深入研究设计；
用户的行为明细数据存放位置有如下可选方案

• 所要计算的明细数据，全部存放在外部存储系统，比如doris中
• 所要计算的明细数据，全部存放在规则引擎的内部存储，比如flink的state中

2）明细数据存储的两难选择

选择1 - 全部存储在doris中

在规则引擎中高并发高频度地去请求doris进行查询
优势
doris有着强大的查询功能，这样一来，不论规则中的判断条件有多复杂，我们在进行计算实现时，对不同的行为判断，无非是换一个sql而已，而这个sql还可以由规则发布平台在发布规则的时候直接生成并跟随规则内容一起注入到规则引擎系统；

弊端

• doris无法承受如此高并发、高频度、复杂查询，因而不能满足我们极为严苛的响应延迟要求；
• 一个最新的行为事件，到达flink算子需要进行计算判断时，它是否已经存在于doris中 如计算A的次数是否满足，当A到达flink时，如果这个A尚未存在于doris，则计算将有遗漏 暂时无法在飞书文档外展示此内容 这要求doris有强大的实时数据高频写入能力（而从doris的底层架构来看，它并不擅长）；

选择2 - 全部存储内部state中

在规则引擎中，直接基于内部state的明细数据进行行为条件计算判断

优势

• 不需要查询外部系统，减少系统对外部的依赖，降低了系统的架构复杂度、运维复杂度
• 内部state中的进行运算，速度更有保证

弊端

• 1）由于我们要求支撑的规则行为条件，可以跨很长的时间窗口，这样就导致state中需要存储很长时间范围的用户明细数据，这对于state来说，checkpoint将变得无法完成，从而导致整个flink程序必将崩溃
• 2）在state中存储用户明细，对这些明细数据的复杂计算，无法利用doris那样的sql查询能力，一切都得使用硬编码来实现，且不说各种复杂计算下的硬编码开发难度，更致命的是，这将导致我们的flink程序无法做到用一套逻辑代码来支撑各种各样的规则条件计算需求

复杂判断条件的编码问题
从上文可以看出，方案中存在着对上线后数据进行编程计算的需求
而对数据的计算逻辑，有简单的，也有复杂的，而且各种条件的计算逻辑各不相同，这对我们编程带来的最大难度是，无法用同一套代码来应付灵活多变的规则条件计算逻辑；
虽然，我们可以根据当前的需求进行抽象总结，将各类需求的计算逻辑进行分门别类，从而设计出N套枚举式的硬编码，但这样一来，灵活性肯定不够，如果要新上线一个超出此前枚举逻辑的新规则，那就不得不关闭系统，重新编码，打包上线；
这显然不符合一开始的构想： 可以在线注入新规则（即动态规则）

==========================================================================

3.2.3 方案概要设计1（即时画像条件计算的数据存储问题）

既然把长时间跨度的明细数据存放在flink state中是完全不可行的，那么，我们首先排除这种方案；
既然无法用一种简单粗暴的方案来解决全场景需求，那么，我们可以针对具体场景做具体设计；

1）只涉及上线前的行为分析判断（历史）
既然只涉及历史数据，一个用户是否满足条件不会随时间而改变，只需要查询判断一次即可；
这样，我们完全可以仿照前面的 “画像条件判断实现方案” ，在发布规则的时候，从doris中用sql进行统计，并将统计结果形成固化的结果数据，接入规则引擎或从外部提供给规则引擎取用；

2）只涉及上线后的行为分析判断（未来）
可以基于flink state来计算；
请注意： 上面说的是 基于flink state来计算 ， 并不是说用flink state来存储所有明细

一个规则上线后的生命周期可能也是很长的，如果从上线时起，state中就不断保存用户的行为明细，这体量也将不可接受；
如果不存储所有明细，那将如何进行行为分析判断呢？
其实，我们在计算一个行为规则条件时，并不一定需要完整的明细，因为：

• 比如，求A事件的次数，其实可以用滚动聚合的方式实现，那样只要记录一个（累计值变量）即可
• 比如，求A事件与B事件的时间间隔，也可以用滚动聚合的方式实现，记录（A最新发生时间，B最早发生时间）即可
• 比如，求A-C-X的序列，用滚动聚合的方式实现，记录 List[A，C，… ]即可
• 比如，求A的某属性统计值，用滚动聚合的方式实现，记录一个（累计值变量）即可

综上，从理论上来说，对于只涉及上线后的行为分析计算，即使不利用state保存所有明细数据，依然可以实现，而是利用滚动运算方式，只要存储滚动运算的聚合值即可

3） 横跨上线前后时间段的行为分析判断（历史+未来）
这是最麻烦的一种，最省事的办法是从需求层面去规避，不提供此类需求的支撑即可
由于时间窗口跨上线前后，就导致既无法仅查询历史数据，也不能仅滚动聚合上线后的数据
唯一的办法是：

1. 将规则条件时间窗口，分解成上线前和上线后两段；
2. 上线前的历史数据的查询结果，传递给规则引擎，然后规则引擎将这个结果和后续的数据分析进行接续整合（把历史查询结果，作为未来查询运算时的初始值）

这里的数据结构设计，就比较复杂，需要具体设计

• 单事件（次数）判断 比如，条件是：要求发生过事件A次数>=5 有可能在历史数据中查到了5+次，则已经符合要求，不需要再判断上线后了 有可能在历史数据中查到了3次，就要求在上线后继续判断是否再发生2+次

对于历史查询已满足条件的人，可以直接整合到满足画像的人群固化数据中去；
那么，历史查询结果需要传递给规则引擎的数据该怎么组织呢？

用户：规则：条件：值
用户_1:规则_x:条件_1:3

• 单事件属性统计值判断 比如，条件是：要求发生过事件A的sum(某属性)>=5 有可能在历史数据中查到已经符合要求，不需要再判断上线后了 有可能在历史数据中查到的结果为4，就要求在上线后继续统计

对于历史查询已满足条件的人，直接整合到满足画像的人群固化数据中去；
那么，历史查询结果需要传递给规则引擎的数据该怎么组织呢？

用户：规则：条件：值

• 事件间隔关系判断 比如： 要求发生过：事件A和时间B间隔>5（+ 这个模式的发生次数>2），如何分解呢？

有可能在历史数据中查询中，已经发生过，则已经符合要求，上线后不需要判断了
有可能在历史数据中，发生过了A，也甚至发生过B，只是间隔不满足，那么，可以记下没有后续B事件的最晚A，等上线后再看是否有B

对于历史查询已满足条件的人，直接整合到满足画像的人群固化数据中去；
那么，历史查询结果需要传递给规则引擎的数据该怎么组织呢？
用户：规则：条件
满足次数：0
待满足序列[A:时间, ]

• 事件序列关系判断 比如： 要求依次发生过 A - C - E 行为序列（+发生次数>2），又该如何分解呢？ 有可能在历史数据中查询到了 A E C ，那可认为已满足了 A - C （步骤2），待上线后等待 E 即可 有可能在历史数据中1个步骤都没有，则全待上线后进行分析判断 对于历史查询已满足条件的人，直接整合到满足画像的人群固化数据中去； 那么，历史查询结果需要传递给规则引擎的数据该怎么组织呢？

用户：规则：条件
满足次数：0
待满足序列[A ， C  ，  ]

-- 也可以这样： 传递用户所做过的行为序列字符串
满足次数：2
待满足序列 "AEC"
暂时无法在飞书文档外展示此内容

• 事件属性统计值关系判断 比如：要求事件A的 sum(某属性值）> 事件B的sum(某属性值） 对于历史查询已满足条件的人，直接整合到满足画像的人群固化数据中去； 那么，历史查询结果需要传递给规则引擎的数据该怎么组织呢？

用户：规则：条件
A：sum(某属性值）
B：sum(某属性值）

4）条件运算过程中的滚动聚合中间结果存储最终方案
综上所述，各种条件的滚动增量运算逻辑不同，所需要记录的中间聚合结果数据结构也不同，还需要将历史时段的条件查询结果和未来的在线运算状态初始值进行整合，最终选择的“滚动运算中间结果存储方案”为：
将中间结果存储在redis中

• redis对外开放，需要对运算结果或初始值等进行人为干预的话，很方便
• redis中有丰富的数据结构，将为我们的滚动增量运算逻辑设计带来巨大的潜在的帮助
• redis本身的读写速度与flink内部的state不相上下，能满足我们的高并发低延迟读写要求

3.2.4 方案概要设计2（即时画像条件计算逻辑的硬编码问题）

综上所述，规则条件的查询计算，需要有sql和代码这两种具体实现
1）两种计算需求

1. 在引擎外，在规则上线前，用sql在doris的历史明细数据中进行统计（这个好实现）
2. 在引擎内，在规则上线后，对流入的行为事件进行统计、判断（这个需要api编程），甚至还要接续历史查询结果进行 “现时态” 数据的整合计算

对于第2种需求，显然不是简单的逻辑代码
而且，这个编程还不能简单使用硬编码（或者多套枚举式硬编码）
那该怎么办呢？

2）规则条件计算核心设计

1. 合理设计好历史数据查询结果的数据结构及存储方式、及传递机制
2. 引入动态脚本语言groovy，来实现具体的计算逻辑

上文中分析的结论：

• 各种条件的历史查询结果形式多样，基本上无法统一成一种数据结构模式
• 并且，不同条件的历史查询结果和后续的“现时态”数据整合计算逻辑，也无法统一成一种模式

而这些 灵活多变的规则条件的 “ 结果数据结构和运算逻辑 ” ，还要尽可能与我们flink规则引擎的通用逻辑进行剥离；以便于在不对规则引擎系统作任何改变的情况下，依然可以支撑 各类规则条件的 “ 结果数据结构和计算逻辑 ” ；
也就是说，规则条件的具体计算逻辑，对于flink规则引擎的通用处理流程来说，需要是透明的！

首先，借鉴面向对象的思想

1）高内聚
把 规则条件计算过程中的状态数据（历史查询结果数据、 滚动聚合的中间结果） 和 计算逻辑（包含滚动聚合逻辑和历史数据接续整合逻辑） ，进行高度绑定内聚！形成一个高内聚的“状态机”
每一类规则条件的计算，都有一个自己的 规则运算状态机 ；
每一种“状态机”，都封装了自己的计算逻辑，也负责维护、理解自己的数据结构；

2）低耦合
当新增规则时，可以由外部系统为规则去选用一个 “运算机”，然后注入到我们的规则引擎，并由规则引擎去驱动这些运算机进行运算即可，这样，我们的规则引擎就跟规则条件运算状态机实现了解耦

其次，将 状态机的 运算逻辑 动态化

1）状态机 的运算逻辑，用动态脚本语言来实现，如groovy
2）规则条件状态机的逻辑代码，将在新规则发布上线时注入到规则引擎中

最后，将状态机的 状态数据 存储外部化

状态机的状态数据，本可以直接存储在规则引擎内部（flink的state中）；
但考虑到如下因素：

• 状态数据的可维护性（比如在特定场景下需要人为去更新维护，或者需要重置等）、通用性、可全局共用性等
• 以及考虑到减轻flink的state维护压力、checkpoint压力等

我们考虑将状态机的状态数据存储在外部的redis集群中

• 这不光可以增加规则运算的灵活性，运维的便捷性
• 更能充分利用redis数据结构功能来简化我们自己的编程复杂度和运行效率

4 项目完整架构
4.1 技术架构图
通过上面的设计，我们形成了如下的最终架构设计图

4.2 架构概要说明

1. 用户画像标签库，由离线数仓组提供，并将标签数据存储在elastic search中，方便画像人群圈选
2. 用户行为事件明细数据（最近3个月），实时落地存入doris数据库中，方便快速统计历史时间窗口中的规则条件
3. 开发规则管理web系统，功能主要有：
4. 为营销人员提供规则定义、管理等操作的可视化平台，并能根据用户定义的规则条件，选择运算逻辑groovy脚本模板
5. 可以根据规则条件的定义，查询doris历史明细，形成规则条件计算的初始化状态数据，并发布到状态数据redis集群
6. 可以根据规则条件发布或人为触发，进行人群圈选，并形成bitmap，并存入mysql元数据库
7. 可以将规则定义的相关参数和计算代码groovy脚本，存入mysql元数据库
8. 开发人员，可以往规则管理平台中维护、添加groovy脚本模板，并开发新的规则定义页面，为营销人员提供的新的规则模板；
9. 开发flink流式处理系统（规则运算引擎），消费kafka中的用户行为明细数据，并通过cdc-connector接受规则管理系统发布的规则信息，对不断流入的用户行为数据，驱动各规则的运算状态机，进行各类规则的条件运算、判断，并输出最终触达结果