Flink cep notFllowedBy 用法案例

今天药忘吃喽~ 2021-11-17 10:32 243阅读 0赞

notFllowedBy 意思是只保留第一个条件最后的一条数据跟第二个条件去匹配

<table> 
 <tbody> 
  <tr> 
   <td>notNext()<br> notFollowedBy()</td> 
   <td>后面的模式不命中（严格/非严格）</td> 
  </tr> 
 </tbody> 
</table>

下图代码说明的就是，匹配最后一个 name = '2',，接下来遇到 '0'  就是匹配结果，所有会有2个结果

(Event(A,a3,2) ,Event(A,a6,0))  
(-Event(A,a8,2),Event(A,a10,0))

package com.coder.flink.restart
    
    import org.apache.flink.cep.scala.{CEP, PatternStream}
    import org.apache.flink.cep.scala.pattern.Pattern
    import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.scala._
    import scala.collection.Map
    
    object demo1 {
      case class Event(types: String, temp: String, name: String)
    
      def main(args: Array[String]): Unit = {
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val data: DataStream[Event] = env.fromElements(
    
          Event("A", "a1", "2"),
          Event("A", "a2", "2"),
          Event("A", "a3", "2"),
          Event("A", "a4", "1"),
          Event("A", "a5", "1"),
          Event("A", "a6", "0"),
          Event("A", "a7", "3"),
          Event("A", "a8", "2"),
          Event("A", "a9", "1"),
          Event("A", "a10", "0")
        )
    
        //定义Pattern接口
        val pattern = Pattern.begin[Event]("start") //指定名称
          .where(_.name.equals("2"))
          .notFollowedBy("middle") //接下来的名称
          .subtype(classOf[Event]) //可以将Event事件转换TempEvent事件
          .where(_.name.endsWith("2"))
          .followedBy("last").where(_.name.endsWith("0"))
    
        val patternStream: PatternStream[Event] = CEP.pattern(data, pattern)
    
        //获取触发事件
        val reslut: DataStream[List[Event]] = patternStream.select((pat: Map[String, Iterable[Event]]) => {
          val rs: List[Event] = pat.get("last").get.toList
          val rs2: List[Event] = pat.get("start").get.toList
          val aa: Event = rs(0)
          val aa2: Event = rs2(0)
          println(aa +"----"+aa2)
          rs
        })
    
    
    
        reslut.print()
        //    data.print()
        env.execute()
      }
    }

输出结果为：

Event(A,a6,0)----Event(A,a3,2)  
Event(A,a10,0)----Event(A,a8,2)  
2> List(Event(A,a10,0))  
1> List(Event(A,a6,0))