Flink CEP 示例（可运行）

package com.cep

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.pattern.conditions.SimpleCondition
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.scala.KeyedStream
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.util.Properties
import java.util

object Testcep {
  def main(args: Array[String]){
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val initProps = new Properties()
    initProps.setProperty("bootstrap.servers", "cwh66:9092")
    initProps.setProperty("group.id", "test22")

    // 定义一个cep pattern模式。此处复杂的事务为：用户click后马上进行buy操作
    val pattern: Pattern[UserAction, UserAction] =
      Pattern.begin[UserAction]("start").where(new SimpleCondition[UserAction] {
        override def filter(t: UserAction): Boolean = t.action.trim.equals("click")
      })
        .next("middle").where(new SimpleCondition[UserAction] {
        override def filter(t: UserAction): Boolean = t.action.trim.equals("buy")
      })

    // 获取一个普通的流
    val input: KeyedStream[UserAction, String] = env.addSource(
      new FlinkKafkaConsumer[String]("test2", new SimpleStringSchema(), initProps))
      .map { line =>
        val strs: Array[String] = line.split(",")
        UserAction(strs(0), strs(1).toLong, strs(2), strs(3)) // 将记录转换为UserAction类型
      }.keyBy(_.name)

    // 将我们定义好的cep pattern应用于这个普通的流
    val patternStream: PatternStream[UserAction] = CEP.pattern(input, pattern).inProcessingTime() or inEventTime()
     //注意必须添加inProcessingTime() or inEventTime()，否则程序不能输出数据

    // 通过select算子获取符合pattern的事务数据，并打印结果
    val patternResult= patternStream.select(new PatternSelectFunction[UserAction, String] {
      override def select(map: util.Map[String, util.List[UserAction]]): String = {
        val click: UserAction = map.get("start").iterator().next()
        val buy: UserAction = map.get("middle").iterator().next()
        // 打印用户的名称，点击和购买的时间
        s"name: ${click.name}, click: ${click.timestamp}, buy: ${buy.timestamp}"
      }
    })

    patternResult.print()

    env.execute("click buy test")

  }


}
case class UserAction(ip: String, timestamp: Long,name: String,  action: String)