在这样一种场景，用户的登录行为数据都会以LoginEvent的行式记录下来，每次失败或者成功以及错误都会记录下来，一般客户端都会进行检验，正常的用户不可能在一秒钟之内登录错误多次，这时候我就得怀疑这部分数据是不是机器对用户的密码进行暴力破解，如果有需要我们得将这些攻击IP进行封锁。

Flink - CEP 优点

复杂性：多个流join，窗口聚合，事件序列或patterns检测
低延迟：秒或毫秒级别，比如做信用卡盗刷检测，或攻击检测
高吞吐：每秒上万条消息

执行流程

Flink-CEP监控

用户的登录日志数据会以实时的方式传递给Flink，常用的有Kafka,MQ等消息中间件。

接着使用Flink-CEP进行模式匹配，匹配到了就会发出告警处理。

案例

用户登录日志格式如下表

依赖

compile group: 'org.apache.flink', name: 'flink-streaming-scala_2.11', version: "1.6.2"
compile group: 'org.apache.flink', name: 'flink-cep-scala_2.11', version: "1.6.2"

使用流式处理环境，并模拟上面的登录日志

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<LoginEvent> loginEventStream = env.fromCollection(Arrays.asList(
                new LoginEvent("1","192.168.0.1","fail"),
                new LoginEvent("1","192.168.0.1","fail"),
                new LoginEvent("1","192.168.0.1","fail"),
                new LoginEvent("2","192.168.10,10","success")
        ));

定义一个登录日志的POJO

public static class LoginEvent implements Serializable {
        private String userId;//用户ID
        private String ip;//登录IP
        private String type;//登录类型

        public LoginEvent() {
        }

        public LoginEvent(String userId, String ip, String type) {
            this.userId = userId;
            this.ip = ip;
            this.type = type;
        }
        // gets sets
}

再定义一个告警的POJO

public static class LoginWarning implements Serializable {
        private String userId;
        private String type;
        private String ip;

        public LoginWarning() {
        }

        public LoginWarning(String userId, String type, String ip) {
            this.userId = userId;
            this.type = type;
            this.ip = ip;
        }
}

开始定义匹配模式,首先为第条日志定义一个first名称，如果满足第一个where条件，则进入下一个监听事件second,如果在 1秒 钟之内两个模式都满足，则成为loginFail告警。

Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>
                begin("begin")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context context) throws Exception {
                        return loginEvent.getType().equals("fail");
                    }
                })
                .next("next")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context context) throws Exception {
                        return loginEvent.getType().equals("fail");
                    }
                })
                .within(Time.seconds(1));

登录失败事件模式定义好了之后，我们就可以把它应用到数据流当中了。

因为我们是针对用户这个维度进行监听的，所以我们需要对用户进行分组，以便可以锁定用户IP。

PatternStream<LoginEvent> patternStream = CEP.pattern(
                loginEventStream.keyBy(LoginEvent::getUserId),
                loginFailPattern);

所匹配的到事件会以一个Map<String, List<LoginEvent>>返回,key为事件名称，value为匹配的数据列表。

DataStream<LoginWarning> loginFailDataStream = patternStream.select((Map<String, List<LoginEvent>> pattern) -> {
            List<LoginEvent> first = pattern.get("begin");
            List<LoginEvent> second = pattern.get("next");

我们可以将告警事件直接打印出来

loginFailDataStream.print();

最终会产生如下两条告警

6> LoginWarning{userId='1', type='192.168.0.2', ip='fail'}
6> LoginWarning{userId='1', type='192.168.0.3', ip='fail'}

完整项目代码 github传送门

附送Scala版本

Scala 版本就是那么简洁明了

object FlinkLoginFail {

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

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val loginEventStream = env.fromCollection(List(
      new LoginEvent("1", "192.168.0.1", "fail"),
      new LoginEvent("1", "192.168.0.2", "fail"),
      new LoginEvent("1", "192.168.0.3", "fail"),
      new LoginEvent("2", "192.168.10,10", "success")
    ))

    val loginFailPattern = Pattern.begin[LoginEvent]("begin")
      .where(_.getType.equals("fail"))
      .next("next")
      .where(_.getType.equals("fail"))
      .within(Time.seconds(1))

    val patternStream = CEP.pattern(loginEventStream, loginFailPattern)
    
    val loginFailDataStream = patternStream
      .select((pattern: Map[String, Iterable[LoginEvent]]) => {
        val first = pattern.getOrElse("begin", null).iterator.next()
        val second = pattern.getOrElse("next", null).iterator.next()

        new LoginWarning(second.getUserId, second.getIp, second.getType)
      })

    loginFailDataStream.print

    env.execute
  }

}