Flink基礎(十七): Flink CEP 簡介
阿新 • • 發佈:2020-08-11
1.什麼是 CEP
概念:
一個或多個由簡單事件構成的事件流通過一定的規則匹配,然後輸出使用者想得到的資料,滿足規則的複雜事件。 特徵: 目標:從有序的簡單事件流中發現一些高階特徵 輸入:一個或多個由簡單事件構成的事件流 處理:識別簡單事件之間的內在聯絡,多個符合一定規則的簡單事件構成複雜事件 輸出:滿足規則的複雜事件 CEP 用於分析低延遲、頻繁產生的不同來源的事件流。CEP 可以幫助在複雜的、不相關的事件流中找出有意義的模式和複雜的關係,以接近實時或準實時的獲得通知並阻止一些行為。 CEP 支援在流上進行模式匹配,根據模式的條件不同,分為連續的條件或不連續的條件;模式的條件允許有時間的限制,當在條件範圍內沒有達到滿足的條件時, 會導致模式匹配超時。 看起來很簡單,但是它有很多不同的功能: 輸入的流資料,儘快產生結果 在 2 個 event 流上,基於時間進行聚合類的計算 提供實時/準實時的警告和通知 在多樣的資料來源中產生關聯並分析模式 高吞吐、低延遲的處理 市場上有多種 CEP 的解決方案,例如 Spark、Samza、Beam 等,但他們都沒有提供專門的 library 支援。但是 Flink 提供了專門的 CEP library。2 Flink CEP
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-cep_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency>
case class LoginEvent(userId: String, ip: String, eventType: String, eventTime: String)
val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) env.setParallelism(1)
val loginEventStream = env.fromCollection(List( LoginEvent("1", "192.168.0.1", "fail", "1558430842"), LoginEvent("1", "192.168.0.2", "fail", "1558430843"), LoginEvent("1", "192.168.0.3", "fail", "1558430844"), LoginEvent("2", "192.168.10.10", "success", "1558430845") )).assignAscendingTimestamps(_.eventTime.toLong)
3Pattern API
4 個體模式
個體模式的條件