如何利用Flink实现超大规模用户行为分析

各位晚上好，首先感谢大家参与我的这次主题分享，同时也感谢 InfoQ AI 前线组织这次瀚思科技主题月！

瀚思科技成立于 2014 年，按行业划分我们是一家安全公司。但和大家熟知的卖杀毒软件或者防火墙的传统安全公司不同。瀚思科技帮助各种中大型企业搭建安全大数据的分析平台，平台上应用的安全分析策略深度结合了多种机器学习算法，最终帮助企业定位与揭示各种安全问题。所以我们自己定位是一家安全 + 大数据 +AI 的公司。

言归正传，今天的分享主题是：基于 Flink 流处理的动态实时大规模用户行为分析

今天的分享主要包括四大部分：

1）网络安全中的用户行为分析（简称 UBA）；

2）实时超大规模用户行为分析的技术挑战 ；

3）Drools 规则引擎在 CEP 中的应用 ；

4）Flink 原生 CEP 组件。

首先，我们先明确一个概念，什么是网络安全中的用户行为分析？简而言之，用户行为分析是通过分析用户数据（例如交易数据，用户登录数据），找出异常行为以检测外部及内部人士的攻击活动。举例来说，外部攻击通常是由外部黑客通过破解 VPN 密码并夺取员工帐户的方式实现。而内部攻击则往往表现为心存不满的或者即将离职的员工对敏感信息的窃取。从右图可以看出，我们需要分析的源数据通常表现为多种数据类型，例如服务器数据、网络数据、数据库数据、应用程序数据、安全数据等。这些数据被送入用户行为分析系统，输出用户正常行为和异常行为（内 / 外部攻击）。

传统的用户行为分析系统通常以离线批处理模式根据既定规则对这些数据进行分析。比如每天定时跑前一天的数据。这种方式适合对实时性要求不高甚至没有要求的业务场景。而面对强调实时性的场景，例如反欺诈场景，需要对实时交易数据进行分析，及时应对不正常的交易。在这种情况下，离线的方式显然不适用。我们必须利用在线 / 流式处理框架，并添加必要的机器学习算法，对实时数据进行 7/24 监控，以区分威胁行为与正常行为。

然而大规模用户行为分析并没有想象中那么简单，尤其是对实时性要求高的场景下。实际应用中，会存在很多细节问题以及技术挑战。我们可以看到这里列举出了四点主要的技术挑战。第一点，输入信息规模过大。举例来说：在一个有 10k+ 员工的大型企业部署用户行为分析系统时，往往需要对这 10k+ 用户在十余个维度上做分析。面对这样体量的数据。分析性能往往受到巨大的挑战。第二点也就是我们先前谈到的实际需求往往需要以实时方式检测攻击活动。第三点讲的是检测逻辑往往不是单一的，实际应用中需要将黑名单、业务逻辑规则以及机器学习算法加以结合。一旦规则数量过大，单个规则逻辑复杂时，整个系统的吞吐量就会受到挑战。第四点讲的是规则不是静态的，而是需要被实时更新的，也就是规则的动态部署。

为了应对以上问题，我们通过技术调研选择了 Flink 作为底层的流处理框架。主要出于以下几点原因：

第一，Flink 是一个纯流式系统，吞吐量实际测试可达 100K EPS。而不像某些框架是用 mini batch 的模式来达到所谓的流式处理的；

第二，面对不同的用户数据格式，我们必须支持多种数据源，这一点上 Flink 内置的对多种数据源的支持（CSV，Kafka，Hbase，Text，Socket 数据等）也为用户数据的接入提供了便利；

第三，Flink 强大的窗口机制（包括翻转窗口，滑动窗口，session 窗口，全窗口以及允许用户自定义窗口）可以满足复杂的业务逻辑，使得用户可以编写复杂的业务规则；

第四，Flink 内置的 RocksDB 数据存储格式使其数据处速度快且资源消耗少，在 Checkpoint 上起到了至关重要的作用；

第五，Flink 对算子（operator）的高可控性，使得用户可以灵活添加删除或更改算子行为。这一点对于动态部署有着至关重要的意义。

下面我们来看一下整个系统的架构。上图中左边为数据源，右边为监测结果输出，我们在之前已经提过。中间是整个系统的核心部分。可以看到，底层的 Flink 是整个实时系统的根基，上方绿色部分为三种 ETL 类型：统计指标、实体关系与序列。我们可以将这些 ETL 类型转换为由 Drools 规则引擎解析并配合机器学习算法的 Scenario 规则。同时对于规则和告警事件，我们需要提供良好的交互界面。此图中省去了数据源输入所依赖的一些技术，比如 Kafka，Elasticsearch。而每一条数据在进入 Flink 时会按 user key 做 partition。同时我们根据场景特点对 Drool 规则引擎做了必要的修改，定义了规则所包含的一些基本概念。这一点会在下面具体展开。

规则引擎方面我们有两个选择：Flink 原生 CEP 组件和 Drools 规则引擎。那么两者各有什么优势和劣势呢？首先我们看一下 Flink CEP。当前稳定的 Flink1.3 版本的 CEP 是一套极具通用性、易于使用的实时流式事件处理方案。作为 Flink 的原生组件，省去了第三方库与 Flink 配合使用时可能会导致的各种问题。但其功能现阶段看来还比较基础，不能表达复杂的业务场景，同时它不能够做到动态更新（这是一个痛点）。具体如何解决我们稍后会看到。

接下来我们看看什么是 Drools。Drools 是一套基于 JVM 的，实现了 RETE 算法的规则引擎。它可以将多变的规则从硬编码中解放出来，以规则脚本的形式存在。右边图中显示的是一个典型的 Drools 规则的定义方式。可以看到，其语义与 Java 非常类似。既可以导入既有的 Java POJO（图中 Person 类），也可以在规则文件中直接定义类（EventA）。when 语句中是具体的判断条件，then 语句中是满足判断条件之后所做的操作。操作可以是任意的，不仅限于对满足条件的那个对象进行操作。比如你可以在 then 里调用某个 Java 类的方法，或者调用某个全局变量。总之，可以在 Drools 规则文件中 import Java 类，然后对其进行操作。

那么 Drools 有些什么优缺点呢？它最大优势在于语法规则简单，类似 Java，编写门槛不高、能够无缝化与 Java 集成，且用户可以对 Drools 规则进行动态配置。但这套方案也存在着自己的不足之处：例如其内置聚合功能速度缓慢，不适合我们自身或者客户使用场景下的大量聚合操作任务。另外，其内置事件序列处理机制也需要消耗大量内存资源。

具体如何攻克这些缺点，待会儿我们会提到。接下来我们先看看如何将业务场景转变成 Drools 规则。

作为常用的业务场景，我们需要将三种 ETL 类型翻译成对应的 Drools 规则。具体来讲，由 Flink 对源数据进行预处理生成的事件数据中的每一行都需要由三种 ETL 类型进行处理：统计指标、实体关系与序列，并借此将内容转换为实际行为。统计指标是指对某个或某几个字段在特定窗口内做聚合，例如一小时内的某用户的登录次数或者某个时间段内连续登录失败的次数。实体关系关注的是两个实体之间的关联关系。实体可以是用户，部门，设备，邮件，地理位置等。那么哪个用户使用了哪台设备就是一个实体关系。哪封邮件发给了哪个用户也是一个实体关系。

下面我们来看一个具体的例子。

可以看到我们这里有一条检测 VPN 可疑行为的规则。规则当中包含三条判断条件。第一条 metric 用来判断一小时能登录失败的次数。第二条演示的是用户与设备之间的实体关系，表达式 expression == “[vpn.user, vpn.device]”说明了这一点。第三条演示的是在序列算法下异常值大于 50 的行为。最后会将满足条件的三个行为收集起来发送给下游的模块。下游模块可以是另一个算子，或者是持久化结果的 DB。

有了 Flink 作为流计算引擎，有了 Drools 作为规则引擎，那么我们如何将两者结合放到一个系统里发挥作用呢。我们需要做的是将源数据输入到 Flink 生成所谓的事件流，同时将 Drools 规则文本读取到 Flink 生成所谓的规则流。而 Flink 中提供了一个 CoFlatMapFunction 可以将两个流结合起来进行分析。在这个 function 里我们所要做的就是将在 Flink 里结合机器学习算法计算出来的结果与 Drools 规则进行匹配。

但事实上，这个方案在实际运行当中会有一些性能上的问题。这些问题主要表现在长周期行为的分析上。比如，机器学习算法需要对长周期行为（数据往往跨越三个月）进行计算，得出异常值。那么这种情况下我们需要维护算法生成的长周期行为的状态。具体方法可以是直接保存在 Drools Engine 中，或者将其保存在外部 DB 中，再或者可以利用 Flink 的 stateful operator 来维护状态。但现有情况下，每种方法都多多少少会有一些问题。接下来我们看看具体问题都有哪些。

需要保存过往窗口的状态，作为中间结果送入 Drools 规则引擎进行计算。Flink 内置的窗口机制在窗口结束时会清除窗口状态。 Flink 内置的 RocksDB 存储结构在窗口清理时会自动删除数据。 Flink 产生的长周期聚合结果被送入 Drool 规则引擎进行匹配的时候往往会消耗大量内存。可以看到，主要的痛点就在于中间结果的维护和资源消耗的问题。面对这些问题我们可以尝试以下的做法。

首先想到的是用 redis，memcached 之类的 KV store 来保存中间结果。但实际测试结果表明，它们的性能赶不上 Flink 的速度。所以在追求高吞吐量的情况下，此方法行不通。其次，可以通过修改 Flink RockDB backend 的源码来解决窗口清理时自动删除数据的问题。同时为了保证过期数据不挤压，需要引入“TTL”（time to live）属性，是的 rocksdb 在超时的时候自动删除过期数据。内存问题主要是由 Drools 引擎引起的。因为每一条事件与规则匹配都会生成一个 Fact，默认情况下 fact 无论是否匹配，Drools 都不会立刻删除它。你必须手动的删除它。但当事件数量过大或者规则数量过大时，即使你手动删除没有匹配的 fact，可能也会出现某一时间段大量 fact 存在于内存中的情况。所以可行的办法是设定阈值来控制内存中允许同时存在的 fact 的数量，同时清理失效的 fact。或者也可以尽量保持规则简单化。复杂的聚合规则交给 Flink 去做。

可以看到，以上方案所产生的性能问题主要在于 Drools。其实除了以上的方案，我们还有一个 Plan B。Flink1.4 Snapshot 版本增加了一些新功能。利用这些新功能，我们可以直接使用 Flink CEP 并做到动态更新。这些功能主要包括：新版本加入了对算子粒度的操作。我们可以 checkpoint 某一个算子的状态。同时 Flink CEP 中新增了 pattern group 的概念。可以将多个规则 pattern 归为同一个 group。这样增加了规则的表达能力。利用这些功能，我们重新设计了一个系统来实现规则的动态更新。下面我们来看一下新设计的工作流程。

简单来讲，整个工作流程就是用户更新规则，新规则被翻译成 Java 源码，然后编译并打包成可执行 jar，这个时候系统将触发 Flink 的 Savepoint，保存当前 operator 的状态，然后 cancel 当前运行的 Flink Job，然后把新生成的 jar 发布到 Flink 上去，同时读取最新的 operator 状态，恢复整个系统的运行。值得提出的一点是，根据规则文件里规则的数量和复杂度。我们可以划分规则生成多个 jar 发布到 Flink 上。这样单个 job 的负载就不至于过高。这种动态生成规则代码的方式扩展性和并发性更出色，不存在单一大负载算子。缺陷在于从 Savepoint 到整个流程恢复会有数秒延迟。

具体实践过程中我们发现，如果一个 pattern 对应一条规则流的话，当 pattern 数量过大，程序初始化时就会内存溢出。那么自然而然就想到多个 pattern 对应一条规则流。这就需要用到新版本中的 GroupPattern 的概念。右边图中可以看到 下方的 patterns 是五个 pattern 的组合。这样就解决了内存溢出的问题。另外提一点，此方案下提到的规则并非 Drools 规则，而是根据 Flink CEP 重新定义的一套语法规则。

接下来会有一个视频 demo，我会对 demo 流程做简要说明。

https://v.qq.com/x/page/g0553z9fa6n.html

首先我们登录系统，大家可以看到这里我已经预置了一条规则。此规则中包含两个判断条件：转账数额大于 19000 和小于 300 的。我们现在发布这条规则到 Flink 来看看它是否生效。大家可以看到，在警告界面已经有满足规则的事件被告出。

回到规则界面，我们先关闭这条规则并新增一条按时间窗口聚合的规则。在 10 秒的翻转窗口中设置 5s 的滑动窗口来计算转账数额的总和，一旦总和大于 100000，就被视为可疑事件被告警。同样我们发布这条规则到 Flink 再去告警界面看是否有满足条件的事件。

很显然新规则已经生效，相关告警已经显示在告警界面。同时第一条规则已失效，大家可以看到没有满足第一条规则的事件被告出。让我们再一次回到规则配置界面，让第一条规则再次生效并发布它。这样我们就有两条规则同时在运行。回到告警界面我们可以看到两条规则都已生效。

问答环节

Q（1）这个规则引擎支持自定义变量或者简单逻辑运算么？ Q（2）drools 的性能指标大概是多少？

A：1、无论是 drools 规则还是我们自定义的 CEP 规则都支持自定义变量或者简单逻辑运算。

2、drools 加入到 Flink 里之后，整体的吞吐量大致在 100k 到 200k EPS。

Q：请问翻转窗口是什么意思？

A：翻转窗口即 Tumbling Window，是指有固定时间大小的窗口。比如 1 分钟的翻转窗口。当 1 分钟时间到的时候。对应的窗口函数会被触发，这个函数里会包含这个窗口内的所有记录。函数执行完后会翻转到下一个时间窗口。两个时间窗口不重叠。

Q: 请问 flink 做实时计算，稳定性如何？

A：Flink 的稳定性很高。即使出现由于性能问题引起的程序 halt，你也可以通过 flink 提供的后台界面操控正在运行的 job。只要当前 job 是做了 checkpoint 的，你都可以停止当前 job，重启后会从最近的一次 checkpoint 运行。

Q：我想了解下瀚思的老师有没有调研过 storm+esper 这种方式。另外 spark structured streaming 和 esper 都支持类 SQL 的规则，感觉更方便一些

A：esper 我们有调研过，但是从我们当前已有的场景规则来看，esper 的吞吐量不能和 Flink 相媲美。且 esper 对与硬件的要求也比 Flink 高，相对来说更加吃内存。至于 spark structured streaming，从它给出的官方文档来看，有些我们需要的场景它是无法支持的。比如多个流的聚合操作。事实上，我们也正在开发一套类似 SQL 语法的规则引擎。

讲师介绍

吴昊，瀚思科技高级软件架构师，毕业于 University of Waterloo 计算机系。

精通领域：大数据处理，大型流式处理系统架构设计