[参赛项目] 达标权益，实时领取：基于Flink SQL实时计算达标客户

[duweike]

项目简述

使用Flink SQL实时计算，能够让达标的用户立刻领取到相应的优惠权益。

背景

在活动推广中，发放优惠券是一种常见的营销策略，在银行等金融行业中，仍然存在达标用户领取优惠券时，需要T+1甚至T+n才能兑换的现象，例如购买理财达到一定金额后，赠送给用户的满减券需要过几天带能领取等，这种现象严重影响了客户的使用感受，因此达标客户在T+0时间点收到优惠券对营销活动效果影响很大。

本项目使用Flink SQL这个易用、高效的实时计算框架，能够让业务分析人员无需编程，即可实时收集和统计达标客户信息，并发放优惠券，实现精准营销。

目标

1）开发过程要简单，业务开发人员只需要SQL，不需要编程。
2）达标客户要包含当日新增客户。
3）业务人员可以自由选择活动上线时间，如今天15号，要求达标客户从月初1号到当前时刻起计算达标的用户。
4）达标或不达标用户可追溯，可解释，若用户不达标，能够追查到原因。

实施方案

实施案例：
每月1号到当前时刻，用户入账金额累计大于5000，则达标，可以领取指定优惠券。
1）使用Flink SQL CDC特性，构建实时用户属性表UserTable，属性表实时了，才能满足当日新开户的用户领到优惠券。
2）使用离线数仓跑批SQL加工每月1号到T+1天的每个用户入账总金额，然后每天翻盘导入维表，供最终实时计算使用，设跑批控制表为RunBatchControlTable，批量聚合表为BatchSumTable。
3）使用Flink SQL对流水表TranHistoryTable分组聚合每个用户每天累计入账总金额，设当日聚合的表为TodaySumTable。
4）到这里就需要把离线的数据和实时的数据做一个汇总，然后判断是否满足达标。以TranHistoryTable为源表作为事件触发的表，以跑批控制表的跑批完成日期来精确控制批量累计和与当天的累积和进行汇总，判断最终是否大于达标金额5000。

整个工作流程如图所示：

技术支持：
a）多流顺序性：开发数据延迟功能，多个数据流发生先后顺序保证，对kafka数据源进行源码开发，实现数据延迟功能，以保证当天开户数据和当天汇总的数据早于最终的汇总逻辑。该延迟功能不能用水印实现，因为实际业务的事件时间往往很不可靠。
b）数据源限速：为了提高稳定性提升，受Spark Streaming启发，对Kafka源码修改，开发完成分区限速功能，避免突发流量触发反压。
c）顺序性保证：无法做到整体有序，但是从kafka上游按照用户id hash分区，可以做到每个用户分区有序。
d）初始化问题：使用离线数据的汇总数据，解决了历史存量数据初始化的问题，同时也减少的实时计算的汇总数据，保证了运行的持续稳定性。

成员介绍

团队名称：zybank
天池昵称：duweike，天天进步~，高源之歌，香辣鸡翅，wuhuiqiang

[SomberOfShadow]

good good study, day day up~

[ReberMusk]

都是离线数据计算，如何保证实时的场景呢？比如用户下单15分钟后未完成订单，需要推送一条激活用户继续付费。

[duweike]

都是离线数据计算，如何保证实时的场景呢？比如用户下单15分钟后未完成订单，需要推送一条激活用户继续付费。

1，这个场景不是都是离线数据，是离线数据和实时数据相结合的。也就是同一个用户的历史数据使用离线，当天数据使用实时的，然后把两者实时的进行汇总。
2，您提的这个场景我们也有遇到过类似的，仔细分析后感觉比较复杂，通过别的方式避免了。初步分析的思路是通过自定义函数保存状态，窗口函数进行触发等方式处理，这里的款难点有两个：a）订单->付款是不同的事件，复杂时间处理；b)只有订单，没有付款事件，也就是没有出发是否超时的事件就要借助窗口、状态等来进行触发和判断是否超时了。

[ChenShuai1981]

直接在下单买理财成功后利用AOP切面调用规则引擎算判断是否达标不就行了？历史的有跑批汇总，当天的实时汇总，二者加起来就行了。

[ChenShuai1981]

直接在下单买理财成功后利用AOP切面调用规则引擎算判断是否达标不就行了？历史的有跑批汇总，当天的实时汇总，二者加起来就行了。

[duweike]

直接在下单买理财成功后利用AOP切面调用规则引擎算判断是否达标不就行了？历史的有跑批汇总，当天的实时汇总，二者加起来就行了。

您提到的规则引擎或者是Flink复杂事件处理CEP都是可以解决的，但是会引入一些稍微有点复杂难度的工具，对于只了解SQL的业务人员可能会比较抗拒类似于CEP这样的高级技能，并且在开发过程中的调试等也是很大的难题。
对于我们平台开发维护的人来说，是想要尽量一个简单的方式让业务人员使用，Flink SQL在我们的实践中是最佳的途径。

[ReberMusk]

针对于 Flink CEP 结合规则引擎(动态更新规则) 实现实时触达或投放，此类业务各位大佬有详细的思路吗？我正在做这方面的技术选型，针对 Flink 还在学习中，希望不吝赐教。有几个例子，比如：

• 订单下单后 N 分钟未支付
• 用户在 A 行为前 24 小时内未发生 B 行为

[duweike]

针对于 Flink CEP 结合规则引擎(动态更新规则) 实现实时触达或投放，此类业务各位大佬有详细的思路吗？我正在做这方面的技术选型，针对 Flink 还在学习中，希望不吝赐教。有几个例子，比如：

• 订单下单后 N 分钟未支付
• 用户在 A 行为前 24 小时内未发生 B 行为

这两个场景我觉得也是比较复杂的，关键点就是判断未发生的事件需要保存所有已发生的事件来判断。
1，如果仅有少量的这两个场景，我仍然建议自定义函数（主要是保存状态数据）和窗口（主要是为了触发计算）来处理。
2，如果类似为题比较多，可能引入别的组件比较方便，比如Redis（key失效策略）、CEP（实际生产中我不建议使用，门槛比较高，我们都是避免使用这个特性）。
3，想要启动马上生效，少量的历史数据，比如你说的24小时，可以使用kafka的特性指定开始时间开始消费历史数据，但这个特性有一个弊端或者发生错误的可能，就是kafka不同分区的数据消费速度不同导致逻辑判断错误，有一些逻辑错误可以通过上游生产者发送数据时，同一个用户必须发送到topic的同一个分区避免一部分。

[ReberMusk]

针对于 Flink CEP 结合规则引擎(动态更新规则) 实现实时触达或投放，此类业务各位大佬有详细的思路吗？我正在做这方面的技术选型，针对 Flink 还在学习中，希望不吝赐教。有几个例子，比如：

• 订单下单后 N 分钟未支付
• 用户在 A 行为前 24 小时内未发生 B 行为

这两个场景我觉得也是比较复杂的，关键点就是判断未发生的事件需要保存所有已发生的事件来判断。 1，如果仅有少量的这两个场景，我仍然建议自定义函数（主要是保存状态数据）和窗口（主要是为了触发计算）来处理。 2，如果类似为题比较多，可能引入别的组件比较方便，比如Redis（key失效策略）、CEP（实际生产中我不建议使用，门槛比较高，我们都是避免使用这个特性）。 3，想要启动马上生效，少量的历史数据，比如你说的24小时，可以使用kafka的特性指定开始时间开始消费历史数据，但这个特性有一个弊端或者发生错误的可能，就是kafka不同分区的数据消费速度不同导致逻辑判断错误，有一些逻辑错误可以通过上游生产者发送数据时，同一个用户必须发送到topic的同一个分区避免一部分。

非常感谢答复。业务场景中我举的例子仅是个例，在互联网运营场景中，智能促销、风控管理等都需要类似于此类的复杂计算，业界的智能运营平台也基本上按自定义化的方式去实现的，因此我也在寻找一个较为完备的解决方案。目前的思路是：

• Kafka 做事件采集，利用 Connector 特性快速对接其他组件
• Flink 做历史数据跑批和实时事件数据汇总，延时场景也有较好的支持
• ES 做数据存储
• Drools 做实时规则更新（不太确定这一块，原 Flink 自带 CEP，是否还有必要引入 Drools，貌似阿里的 Blink 已解决规则动态更新的问题）

[duweike]

针对于 Flink CEP 结合规则引擎(动态更新规则) 实现实时触达或投放，此类业务各位大佬有详细的思路吗？我正在做这方面的技术选型，针对 Flink 还在学习中，希望不吝赐教。有几个例子，比如：

• 订单下单后 N 分钟未支付
• 用户在 A 行为前 24 小时内未发生 B 行为

这两个场景我觉得也是比较复杂的，关键点就是判断未发生的事件需要保存所有已发生的事件来判断。 1，如果仅有少量的这两个场景，我仍然建议自定义函数（主要是保存状态数据）和窗口（主要是为了触发计算）来处理。 2，如果类似为题比较多，可能引入别的组件比较方便，比如Redis（key失效策略）、CEP（实际生产中我不建议使用，门槛比较高，我们都是避免使用这个特性）。 3，想要启动马上生效，少量的历史数据，比如你说的24小时，可以使用kafka的特性指定开始时间开始消费历史数据，但这个特性有一个弊端或者发生错误的可能，就是kafka不同分区的数据消费速度不同导致逻辑判断错误，有一些逻辑错误可以通过上游生产者发送数据时，同一个用户必须发送到topic的同一个分区避免一部分。

非常感谢答复。业务场景中我举的例子仅是个例，在互联网运营场景中，智能促销、风控管理等都需要类似于此类的复杂计算，业界的智能运营平台也基本上按自定义化的方式去实现的，因此我也在寻找一个较为完备的解决方案。目前的思路是：

• Kafka 做事件采集，利用 Connector 特性快速对接其他组件
• Flink 做历史数据跑批和实时事件数据汇总，延时场景也有较好的支持
• ES 做数据存储
• Drools 做实时规则更新（不太确定这一块，原 Flink 自带 CEP，是否还有必要引入 Drools，貌似阿里的 Blink 已解决规则动态更新的问题）

1，所有的数据打到kafka有两种方案，一种是通过topic区分，不同的topic的数据结构相同，同一类业务，我们这边绝大部分是对用的源库Oracle中的表对应一个topic；另一种所有的数据打到同一个topic，提前构造一个上百列的宽表，嵌套json，通过不同的字段标识不同的业务。
2，经过计算之后的数据存储使用ES是合适的，不过原始数据应该也是需要存储的，根据数据量可能需要引入别的存储组件。
3，如果有Drools积累的经验，推荐使用Drools，缺点是又引入了一种技术；如果使用FlinkCEP，好处是在同一个Flink框架下完成的判断逻辑，有状态的计算，并且推荐SQL方式的CEP，缺点可能是技术门槛高。