CEP定義

Flink的Complex Event Processing庫

允許我們在事件流中檢測事件的模式

CEP調(diào)研

Reference調(diào)研了網(wǎng)上所有講解CEP的例子，只有4篇文章是有相對完整的代碼的。

[4]中的代碼已經(jīng)得到驗(yàn)證，分為這么幾個步驟:

?

先理一下flink cep的代碼流程[4]

1.先定義Pattern

?Pattern.begin[X]("start").where(...).next("middle").where(...)?

2.通過CEP.pattern()方法將DataStream轉(zhuǎn)化為PatternStream

? val cepResult: PatternStream[Event] = CEP.pattern(inputDataStream, pattern) ?

3.將符合pattern的數(shù)據(jù)調(diào)用select方法對數(shù)據(jù)進(jìn)行處理

cepResult.select(new PatternSelectFunction[X, String] {
? ? ? override def select(pattern: util.Map[String, util.List[X]]): String = {
? ? ? ?? ?logic code
? ? ? ?? ?""
? ? ? }
}

?

?

最終我完成了調(diào)研文獻(xiàn)中的[3][4][5],完整代碼鏈接是:

https://gitee.com/appleyuchi/Flink_Code/tree/master/flink_cep

Reference:

[1]Apache Flink CEP學(xué)習(xí)總結(jié)(評論區(qū)中說代碼存在一些問題)

[2]Apache FlinkCEP實(shí)現(xiàn)超時狀態(tài)監(jiān)控(沒有意義，超時的情況可以由waterMark進(jìn)行sideOutput處理)

[3]Flink - CEP分析攻擊行為(自己補(bǔ)充完整了)

[4]【Flink】flink cep對于復(fù)雜事件的處理(完成)

[5]Flink cep的初步使用(完成)

?

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎勵來咯，堅持創(chuàng)作打卡瓜分現(xiàn)金大獎

總結(jié)

以上是生活随笔為你收集整理的flink的CEP调研与使用的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。