Wormhole_v0.5 重大发布 | Flink 强势加盟，CEP 新鲜亮相

导读：继 Wormhole 的设计思想介绍和功能介绍之后，相信大家对 Wormhole 已经有了初步的了解。2018 年 7 月 31 日，我们发布了 Wormhole_0.5 新版本，与以往基于 Spark 计算引擎的版本相比，该版本新增了基于 Flink 计算引擎的流式处理功能，主要关注低延迟和 CEP。基于 Flink 计算引擎版本具体内容是什么呢？还请各位看官移步正文~

Wormhole Flink****版****介绍

延迟时间是评判流式处理性能的关键指标之一。Spark 基于弹性分布式数据集（Resilient Distributed Dataset，RDD）进行微批处理，所以 Spark 在流式处理方面，不可避免会存在一些延时，只支持秒级延迟。Flink 基于事件处理，实现了真正的流式计算。与基于 Spark 的流式处理相比，它的延迟更低。Wormhole 通过对 Flink 计算引擎的支持，将延迟降低到毫秒级。

Wormhole Flink 版除了支持 Flink SQL，Lookup SQL，新增了对 CEP 的支持，并且支持三者的混合编排，即一个 Flow 中可以包含多个 Flink SQL，多个 Lookup SQL 和多个 CEP。Flink SQL 与 Spark SQL 用法类似，Spark SQL 和 Lookup SQL 在上一篇 Wormhole 系列文章中已经介绍过，这里将不再赘述，下面我们将重点讲解 CEP。

CEP（复杂事件处理）简介

在传统 DBMS 中，所有的操作都只能在数据落库之后才能进行，这极大地降低了事件处理的实时性。与传统 DBMS 不同，CEP 从流式事件中查找匹配指定模式的事件，对流式事件边获取边处理，整个处理过程都在数据流中进行，无需落地，因此它拥有更低的延迟，即所有输入都将被立刻处理，一旦在流式事件中发现了匹配指定模式的事件集，结果就会立即输出。

正因如此，CEP 引起了广泛的关注，并得到了大量的应用推广，主要体现在运营和运维两方面。在运营方面，CEP 经常被应用于金融产品中，例如，股票市场趋势预测、信用卡诈骗预防等。在运维方面，CEP 被用在基于 RFID 的追踪和监控系统中，例如，检测库房中失窃的物品。当然，CEP 也能通过指定嫌疑人的行为，来检测网络入侵。

Wormhole CEP 是基于 Flink CEP 实现的，并且提供了可视化操作界面，无需编码即可快速实现业务需求。Wormhole CEP 引入了窗口时间（Window Time），窗口策略（Strategy），分组策略（KeyBy），输出格式（Output），筛选规则（Pattern）等概念。下面，我们逐一介绍这些概念的含义及用途。

· Window Time:指在触发了符合 Begin Pattern 的事件记录后的窗口时间，如果 watermark 的 time 超过了触发时间+窗口时间，本次 pattern 结束；

· Strategy:包含 NO_SKIP 和 SKIP_PAST_LAST_EVENT 两种策略，前者对应事件滑动策略，后者对应事件滚动策略，具体区别可以借鉴下面的例子:

事件滑动：a1 a2 a3 a4 …

a2 a3 a4 a5 …

a3 a4 a5 a6 …

事件滚动： a1 a2 a3 a4 …

a5 a6 a7 a8 …

a9 a10 a11 a12…

· KeyBy：指依据事件中的哪个字段来做分区。例如，现在有一条数据，它的 schema 包括 ums_id_, ums_op_, ums_ts_, value1, value2 等几个字段，这里选定 value1 来做分区的依赖字段，那么，与 value1 字段相同的数据将被分配到同一个分组上。CEP 操作将分别针对每一分组的数据进行处理，KeyBy 可以作用在多个字段上。

· Output：输出结果的形式，分为三类：Agg、Detail、FilteredRow

§ Agg:将匹配的多条数据做聚合，生成一条数据输出

例：field1:avg,field2:max（目前支持 max/min/avg/sum）

§ Detail:将匹配的多条数据逐一输出

§ FilteredRow:按条件选择指定的一条数据输出

例：head/last/field1:min/max

· Pattern：筛选规则。每个 CEP 由若干个 Pattern 组成。

每个 Pattern 包括以下三个概念：

§ Operator:操作算子。CEP 中的第一个 Pattern Operator 只能为 begin，其后的每个 Pattern Operator 只能为 next、followedBy、notNext、notFollowedBy 四种类型中的一种，其含义分别为：

✔ next:会追加一个新的 Pattern 对象到既有的 Pattern 之后，它表示当前模式运算符所匹配的事件必须是严格紧邻的，这意味着两个被匹配的事件必须是前后紧邻，中间没有其他元素；

✔ followedBy:会追加一个新的 Pattern 到既有的 Pattern 之后（其实返回的是一个 FollowedByPattern 对象，它是 Pattern 的派生类），它表示当前运算符所匹配的事件不必严格紧邻，这意味着匹配的两个事件之间允许穿插其他事件；

✔ notNext:增加一个新的否定模式。匹配(否定)事件必须直接输出先前的匹配事件(严格紧邻)，以便将部分匹配丢弃；

✔ notFollowedBy:会丢弃或者跳过已匹配的事件(注：notFollowedBy 不能为最后一个 Pattern)。

§ Quantifier:用来指定满足某一 pattern 的事件数量。目前配置包括：一条及以上，指定条数，指定条数及以上；这里需要特殊说明的是，notNext、notFollowedBy 这两种 Operator 无法设置 Quantifier；

§ Conditions:判断条件。用户可以针对事件的某个或多个属性设置判断条件，例如，可以设置只有符合 value1 like a and value2 >=10 的事件才是符合条件的事件。

Wormhole CEP 应用场景

场景一：网络 DDOS 攻击警告

Wormhole CEP 在日常运维中被广泛应用。下面以运维中会遇到的一类情况为例，来介绍如何使用 Wormhole CEP。

DDOS 攻击是日常运维中经常遇到的一类问题，CEP 正好可以用来对 DDOS 攻击进行预警。

DDOS 攻击的判断规则如下：

正常：流量在预设的正常范围内；

警告：某数据中心在 10 秒内连续 2 次上报的流量超过认定的正常值；

报警：某数据中心在 30 秒内连续 2 次匹配警告；

通知：报警后需要短信/邮件通知相关人员。

通过上述规则，DDOS 攻击的判断依据可以被量化为流量超出事件在一定时间内多次产生。只要符合条件，客户请求就可以被认定为 DDOS 攻击。针对符合条件的事件，Wormhole 会向 Kafka 传入报警消息，并由业务系统去 Kafka 中消费报警消息，从而进行相应的后续处理。

1533534473700080275.png

图 1 kafka 业务系统消费示意图

下面，结合一个具体的操作例子来说明 Wormhole CEP 是如何检测 DDOS 攻击的。

首先，针对警告规则，设置一个窗口时间为 10 秒，次数为 2 次，判断条件为流量超过 45(GB)的 CEP，作为第一个 CEP，并将事件发生时间，以及次数 1 作为中间结果进行输出；

1533534490705004541.png

图 2 设置警告 CEP

然后，针对报警规则，再设置一个窗口为 30 秒，判断条件为警告事件发生次数为 2 次作为第二个 CEP。针对符合条件的事件，向 Kafka 中传入报警消息，否则，不做任何操作。

1533534509957015900.png

图 3 设置报警 CEP

最终，设置完两个 CEP 之后，它们将对所有流上事件进行叠加过滤，并针对符合条件的事件，向 Kafka 写入报警消息，从而，协助各个数据中心预防 DDOS 攻击。

1533534575235055211.png

图 4 CEP 列表

场景二：电商业务人工外呼通知

Wormhole CEP 在运营中也起到了重要作用，比如在电商平台中，客户填写提交订单后，由于某些原因长时间未付款，这时需要人工介入处理，如给客户打电话进行回访，从而了解客户情况，提高业务成交量及服务质量。下面以此业务场景为例，介绍如何通过 Wormhole CEP 来实现此类业务需求。

这里将购物步骤简化为两步，第一步提交订单，第二步付款。若某一客户在提交订单后，5min 内未付款，则平台通知工作人员联系客户。假设事件流不断流入 Kafka 中，事件中 userid 字段代表客户 ID；state 字段代表订单状态（s1 是“已提交”，s2 是“已付款”）。通过 CEP 很容易实现上述需求，首先设置第一个 Pattern，用来匹配某客户提交订单事件，即 state=s1；然后设置第二个 Pattern，用来匹配该客户未付款事件，即相邻的事件中 state=s2 未发生。满足两个规则数据即满足需要人工外呼条件，这时系统发消息通知工作人员联系该客户。

1533534617457010085.png

图 5 电商平台数据处理示意图

针对需求，需要设置一个 300s（即 5min）的窗口，然后按照 userid 分组，对每个客户分别进行匹配。

1533534637092087395.png

图 6 CEP 基本配置

第一个 pattern 为客户已提交订单。

1533534664881016572.png

图 7 Pattern Begin

第二个 Pattern 为客户未付款。

1533534678750015487.png

图 8 Pattern notNext

最终，该 CEP 将对所有流上事件进行过滤，并针对符合条件的事件，将数据发送到 Kafka，人工外呼系统根据此数据触发相关业务流程。

总的来说，Wormhole_v0.5 主要是针对 Flink 实现了流式处理，关注点是低延迟和 CEP。目前版本处理支持 Flink SQL，Lookup SQL，CEP，并且支持三者的混合编排。同时，新增 Spark Stream 支持配置用户自定义 Topic，可直接对接 DBus 独立拉全量功能。后续我们会尽快支持 ums_extension(目前支持 ums)、异构 sink（目前支持 Kafka）、udf 等功能。欢迎大家持续关注！

本文转载自宜信技术学院网站。

原文链接：http://college.creditease.cn/detail/162