详解Flink CEP的概念及功能

2021-04-16 11:37

result．print（＂恶意用户＞＞＞＂）
env．execute（＂BarrageBehavior01＂）
｝
｝

实例二：监测刷屏用户

规则：用户如果在10s内，同时连续输入同样一句话超过5次，就认为是恶意刷屏。

使用 Flink CEP检测刷屏用户

object BarrageBehavior02 ｛
case class Message（userId： String， ip： String， msg： String）
def main（args： Array［String］）： Unit ＝｛
／／初始化运行环境
val env ＝ StreamExecutionEnvironment．getExecutionEnvironment
／／设置并行度
env．setParallelism（1）
／／模拟数据源
val loginEventStream： DataStream［Message］＝ env．fromCollection（
List（
Message（＂1＂，＂192．168．0．1＂，＂beijing＂），
Message（＂1＂，＂192．168．0．2＂，＂beijing＂），
Message（＂1＂，＂192．168．0．3＂，＂beijing＂），
Message（＂1＂，＂192．168．0．4＂，＂beijing＂），
Message（＂2＂，＂192．168．10．10＂，＂shanghai＂），
Message（＂3＂，＂192．168．10．10＂，＂beijing＂），
Message（＂3＂，＂192．168．10．11＂，＂beijing＂），
Message（＂4＂，＂192．168．10．10＂，＂beijing＂），
Message（＂5＂，＂192．168．10．11＂，＂shanghai＂），
Message（＂4＂，＂192．168．10．12＂，＂beijing＂），
Message（＂5＂，＂192．168．10．13＂，＂shanghai＂），
Message（＂5＂，＂192．168．10．14＂，＂shanghai＂），
Message（＂5＂，＂192．168．10．15＂，＂beijing＂），
Message（＂6＂，＂192．168．10．16＂，＂beijing＂），
Message（＂6＂，＂192．168．10．17＂，＂beijing＂），
Message（＂6＂，＂192．168．10．18＂，＂beijing＂），
Message（＂5＂，＂192．168．10．18＂，＂shanghai＂），
Message（＂6＂，＂192．168．10．19＂，＂beijing＂），
Message（＂6＂，＂192．168．10．19＂，＂beijing＂），
Message（＂5＂，＂192．168．10．18＂，＂shanghai＂）
）
）
／／定义模式
val loginbeijingPattern ＝ Pattern．begin［Message］（＂start＂）
．where（＿．msg ！＝ null）／／一条登录失败
．times（5）．optional ／／将满足五次的数据配对打印
．within（Time．seconds（10））
／／进行分组匹配
val loginbeijingDataPattern ＝ CEP．pattern（loginEventStream．keyBy（＿．userId）， loginbeijingPattern）
／／查找符合规则的数据
val loginbeijingResult： DataStream［Option［Iterable［Message］］］＝ loginbeijingDataPattern．select（patternSelectFun ＝（pattern： collection．Map［String， Iterable［Message］］）＝＞｛
var loginEventList： Option［Iterable［Message］］＝ null
loginEventList ＝ pattern．get（＂start＂） match ｛
case Some（value）＝＞｛
if （value．toList．map（x ＝＞（x．userId， x．msg））．distinct．size ＝＝ 1）｛
Some（value）
｝ else ｛
None
｝
｝
｝
loginEventList
｝）
／／打印测试
loginbeijingResult．filter（x＝＞x！＝None）．map（x＝＞｛
x match ｛
case Some（value）＝＞ value
｝
｝）．print（）
env．execute（＂BarrageBehavior02）
｝
｝
Flink CEP API

除了案例中介绍的几个API外，我们在介绍下其他的常用API：

1．条件 API

为了让传入事件被模式所接受，给模式指定传入事件必须满足的条件，这些条件由事件本身的属性或者前面匹配过的事件的属性统计量等来设定。比如，事件的某个值大于5，或者大于先前接受事件的某个值的平均值。

可以使用pattern．where（）、pattern．or（）、pattern．until（）方法来指定条件。条件既可以是迭代条件IterativeConditions，也可以是简单条件SimpleConditions。

FlinkCEP支持事件之间的三种临近条件：

next（）：严格的满足条件

示例：模式为begin（＂first＂）．where（＿．name＝＇a＇）．next（＂second＂）．where（．name＝＇b＇）当且仅当数据为a，b时，模式才会被命中。如果数据为a，c，b，由于a的后面跟了c，所以a会被直接丢弃，模式不会命中。

followedBy（）：松散的满足条件

示例：模式为begin（＂first＂）．where（＿．name＝＇a＇）．followedBy（＂second＂）．where（．name＝＇b＇）当且仅当数据为a，b或者为a，c，b，模式均被命中，中间的c会被忽略掉。

followedByAny（）：非确定的松散满足条件

示例：模式为begin（＂first＂）．where（＿．name＝＇a＇）．followedByAny（＂second＂）．where（．name＝＇b＇）当且仅当数据为a，c，b，b时，对于followedBy模式而言命中的为｛a，b｝，对于followedByAny而言会有两次命中｛a，b｝，｛a，b｝。

2．量词 API

还记得我们在上面讲解模式概念时说过的一句话：一般情况下，模式都是单例模式，可以使用量词（Quantifiers）将其转换为循环模式。这里的量词就是指的量词API。

以下这几个量词API，可以将模式指定为循环模式：

pattern．oneOrMore（）：一个给定的事件有一次或多次出现，例如上面提到的b＋。

pattern．times（＃ofTimes）：一个给定类型的事件出现了指定次数，例如4次。

pattern．times（＃fromTimes，＃toTimes）：一个给定类型的事件出现的次数在指定次数范围内，例如2～4次。

可以使用pattern．greedy（）方法将模式变成循环模式，但是不能让一组模式都变成循环模式。greedy：就是尽可能的重复。

使用pattern．optional（）方法将循环模式变成可选的，即可以是循环模式也可以是单个模式。

3．匹配后的跳过策略

所谓的匹配跳过策略，是对多个成功匹配的模式进行筛选。也就是说如果多个匹配成功，可能我不需要这么多，按照匹配策略，过滤下就可以。

Flink中有五种跳过策略：

NO＿SKIP：不过滤，所有可能的匹配都会被发出。

SKIP＿TO＿NEXT：丢弃与开始匹配到的事件相同的事件，发出开始匹配到的事件，即直接跳到下一个模式匹配到的事件，以此类推。

SKIP＿PAST＿LAST＿EVENT：丢弃匹配开始后但结束之前匹配到的事件。

SKIP＿TO＿FIRST［PatternName］：丢弃匹配开始后但在PatternName模式匹配到的第一个事件之前匹配到的事件。

SKIP＿TO＿LAST［PatternName］：丢弃匹配开始后但在PatternName模式匹配到的最后一个事件之前匹配到的事件。

怎么理解上述策略，我们以NO＿SKIP和SKIP＿PAST＿LAST＿EVENT为例讲解下：

在模式为：begin（＂start＂）．where（＿．name＝＇a＇）．oneOrMore（）．followedBy（＂second＂）．where（＿．name＝＇b＇）中，我们输入数据：a，a，a，a，b ，如果是NO＿SKIP策略，即不过滤策略，模式匹配到的是：｛a，b｝，｛a，a，b｝，｛a，a，a，b｝，｛a，a，a，a，b｝；如果是SKIP＿PAST＿LAST＿EVENT策略，即丢弃匹配开始后但结束之前匹配到的事件，模式匹配到的是：｛a，a，a，a，b｝。

Flink CEP 的使用场景

除上述案例场景外，Flink CEP 还广泛用于网络欺诈，故障检测，风险规避，智能营销等领域。

1．实时反作弊和风控

对于电商来说，羊毛党是必不可少的，国内拼多多曾爆出 100 元的无门槛券随便领，当晚被人褥几百亿，对于这种情况肯定是没有做好及时的风控。另外还有就是商家上架商品时通过频繁修改商品的名称和滥用标题来提高搜索关键字的排名、批量注册一批机器账号快速刷单来提高商品的销售量等作弊行为，各种各样的作弊手法也是需要不断的去制定规则去匹配这种行为。

2．实时营销

分析用户在手机 APP 的实时行为，统计用户的活动周期，通过为用户画像来给用户进行推荐。比如用户在登录 APP 后 1 分钟内只浏览了商品没有下单；用户在浏览一个商品后，3 分钟内又去查看其他同类的商品，进行比价行为；用户商品下单后 1 分钟内是否支付了该订单。如果这些数据都可以很好的利用起来，那么就可以给用户推荐浏览过的类似商品，这样可以大大提高购买率。

3．实时网络攻击检测

当下互联网安全形势仍然严峻，网络攻击屡见不鲜且花样众多，这里我们以 DDOS（分布式拒绝服务攻击）产生的流入流量来作为遭受攻击的判断依据。对网络遭受的潜在攻击进行实时检测并给出预警，云服务厂商的多个数据中心会定时向监控中心上报其瞬时流量，如果流量在预设的正常范围内则认为是正常现象，不做任何操作；如果某数据中心在 10 秒内连续 5 次上报的流量超过正常范围的阈值，则触发一条警告的事件；如果某数据中心 30 秒内连续出现 30 次上报的流量超过正常范围的阈值，则触发严重的告警。

Flink CEP 的原理简单介绍

Apache Flink在实现CEP时借鉴了Efficient Pattern Matching over Event Streams论文中NFA的模型，在这篇论文中，还提到了一些优化，我们在这里先跳过，只说下NFA的概念。

在这篇论文中，提到了NFA，也就是Non－determined Finite Automaton，叫做不确定的有限状态机，指的是状态有限，但是每个状态可能被转换成多个状态（不确定）。

非确定有限自动状态机：