十二月综合缴缴情乡村大,久久久久国产精品免费,么公要了我一晚上视频在线观看,日韩电视剧

詳解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，叫做不確定的有限狀態機，指的是狀態有限，但是每個狀態可能被轉換成多個狀態（不確定）。

非確定有限自動狀態機：