基于flink的电商用户行为数据分析【4】| 恶意登录监控

<dependency>
        <groupId>org.apache.flink</groupId>
<artifactId>flink-cep-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>

object LoginFailOne {

  // 输入的登录事件样例类
  case class LoginEvent( userId:Long,ip:String,eventType:String,eventTime:Long)

  // 输出的报警信息样例类
  case class Warning( userId:Long,firstFailTime:Long,lastFailTime:Long,warningMsg:String)

  def main(args: Array[String]): Unit = {

    // 创建流环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    env.setParallelism(1)
    // 设置时间特征为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 读取csv文件
    env.readTextFile("G:\\LoginLog.csv")
       .map(data => {
          // 将文件中的数据封装成样例类
          val dataArray: Array[String] = data.split(",")
          LoginEvent(dataArray(0).toLong, dataArray(1), dataArray(2), dataArray(3).toLong)
        })
        // 设置 WaterMark 水印
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(5)) {
        override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000
      })
      // 以用户id为key，进行分组
      .keyBy(_.userId)
      // 计算出同一个用户2秒内连续登录失败超过2次的报警信息
      .process(new LoginWarning(2))
      .print()

    //  执行程序
    env.execute("login fail job")


  }

  // 自定义处理函数，保留上一次登录失败的事件，并可以注册定时器    [键的类型，输入元素的类型，输出元素的类型]
  class LoginWarning(maxFailTimes:Int) extends KeyedProcessFunction[Long,LoginEvent,Warning]{

    // 定义  保存登录失败事件的状态
    lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState( new ListStateDescriptor[LoginEvent]("loginfail-state", classOf[LoginEvent]) )

    override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {

      // 判断当前登录状态是否为 fail
      if (value.eventType == "fail"){
        // 判断存放失败事件的state是否有值，没有值则创建一个2秒后的定时器
        if (!loginFailState.get().iterator().hasNext){
          // 注册一个定时器，设置在 2秒 之后
          ctx.timerService().registerEventTimeTimer((value.eventTime + 2) * 1000L)
        }
        // 把新的失败事件添加到  state
        loginFailState.add(value)
      }else{
        // 如果登录成功，清空状态重新开始
        loginFailState.clear()
      }
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#OnTimerContext, out: Collector[Warning]): Unit = {
      // 触发定时器的时候，根据状态的失败个数决定是否输出报警
      val allLoginFailEvents: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]()

      val iter: util.Iterator[LoginEvent] = loginFailState.get().iterator()

      // 遍历状态中的数据，将数据存放至 ListBuffer
      while ( iter.hasNext ){
        allLoginFailEvents += iter.next()
        }

      //判断登录失败事件个数，如果大于等于 maxFailTimes ，输出报警信息
      if (allLoginFailEvents.length >= maxFailTimes){
        out.collect(Warning(allLoginFailEvents.head.userId,
          allLoginFailEvents.head.eventTime,
          allLoginFailEvents.last.eventTime,
          "在2秒之内连续登录失败" + allLoginFailEvents.length + "次"))
      }

      // 清空状态
      loginFailState.clear()
    }
  }
}

 // 自定义处理函数，保留上一次登录失败的事件    [键的类型，输入元素的类型，输出元素的类型]
  class LoginWarning(maxFailTimes:Int) extends KeyedProcessFunction[Long, LoginEvent, Warning] {

    // 定义  保存登录失败事件的状态
    lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("loginfail-state", classOf[LoginEvent]))

    override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {
      // 首先按照type做筛选，如果success直接清空，如果fail再做处理
      if(value.eventType == "fail"){
        // 先获取之前失败的事件
        val iter: util.Iterator[LoginEvent] = loginFailState.get().iterator()
        if (iter.hasNext){
          // 如果之前已经有失败的事件，就做判断，如果没有就把当前失败事件保存进state
          val firstFailEvent: LoginEvent = iter.next()
          // 判断两次失败事件间隔小于2秒，输出报警信息
          if (value.eventTime < firstFailEvent.eventTime + 2){
            out.collect(Warning( value.userId,firstFailEvent.eventTime,value.eventTime,"在2秒内连续两次登录失败。"))
          }

          // 更新最近一次的登录失败事件，保存在状态里
          loginFailState.clear()
          loginFailState.add(value)

        }else{
          // 如果是第一次登录失败，之前把当前记录 保存至 state
          loginFailState.add(value)
        }
      }else{
        // 当前登录状态 不为 fail，则直接清除状态
        loginFailState.clear()
      }
    }
  }
  }

object LoginFailWithCep {
  // 输入的登录事件样例类
  case class LoginEvent(userId: Long, ip: String, eventType: String, eventTime: Long)

  // 输出的报警信息样例类
  case class Warning(userId: Long, firstFailTime: Long, lastFailTime: Long, warningMsg: String)

  def main(args: Array[String]): Unit = {
    
    // 1、创建流环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    env.setParallelism(1)
    // 设置时间特征为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 构建数据
    val loginEventStream: KeyedStream[LoginEvent, Long] = env.readTextFile("G:\\idea arc\\BIGDATA\\project\\src\\main\\resources\\LoginLog.csv")
      .map(data => {
        // 将文件中的数据封装成样例类
        val dataArray: Array[String] = data.split(",")
        LoginEvent(dataArray(0).toLong, dataArray(1), dataArray(2), dataArray(3).toLong)
      })
      // 设置水印，防止数据乱序
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000
      })
      // 以用户id为key，进行分组
      .keyBy(_.userId)

    // 定义匹配的模式
    val loginFailPattern: Pattern[LoginEvent, LoginEvent] = Pattern.begin[LoginEvent]("begin")
      .where(_.eventType == "fail")
      .next("next")
      .where(_.eventType == "fail")
      .within(Time.seconds(2))    // 通过 within 关键字定义了检测窗口时间时间

    // 将 pattern 应用到 输入流 上，得到一个 pattern stream
    val patternStream: PatternStream[LoginEvent] = CEP.pattern(loginEventStream,loginFailPattern)

    // 用 select 方法检出 符合模式的事件序列
    val loginFailDataStream: DataStream[Warning] = patternStream.select(new LoginFailMatch())

    // 将匹配到的符合条件的事件打印出来
    loginFailDataStream.print("warning")
    
    // 执行程序
    env.execute("login fail with cep job")

  }

  // 自定义 pattern select function
  // 当检测到定义好的模式序列时就会调用，输出报警信息
  class LoginFailMatch() extends PatternSelectFunction[LoginEvent,Warning]{

    override def select(map: util.Map[String, util.List[LoginEvent]]): Warning = {
      // 从 map 中可以按照模式的名称提取对应的登录失败事件
      val firstFail: LoginEvent = map.get("begin").iterator().next()
      val secondFail: LoginEvent = map.get("next").iterator().next()
         
      Warning( firstFail.userId,firstFail.eventTime,secondFail.eventTime,"在2秒内连续2次登录失败。")
    }
  }
}