Flink实现实时异常登陆监控（两秒内多次登陆失败进行异常行为标记）-阿里云开发者社区

Flink实现实时异常登陆监控（两秒内多次登陆失败进行异常行为标记）

2024-06-12 120

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，5000CU*H 3个月

RDS MySQL Serverless 基础系列，0.5-2RCU 50GB

云数据库 RDS MySQL，集群系列 2核4GB

简介： Flink实现实时异常登陆监控（两秒内多次登陆失败进行异常行为标记）

Flink实现异常登陆监控（两秒内多次登陆失败进行异常行为标记）

在大数据处理领域，Apache Flink 是一个流行的开源流处理框架，能够高效处理实时数据流。在这篇博客中，我们将展示如何使用 Apache Flink 从 MySQL 中读取数据并进行实时异常监控处理，最终将结果写回到 MySQL 数据库中的err_login表中。

项目概述

我们的示例程序将会执行以下任务：

从 MySQL 数据库读取用户登录数据。

过滤出特定状态的登录记录。

对这些记录进行时间窗口处理。

将处理结果写回 MySQL 数据库。

依赖环境

在开始之前，请确保你已经安装了以下环境：

pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>EastMoney</artifactId>
    <version>1.0-SNAPSHOT</version>
    <repositories>
        <repository>
            <id>central</id>
            <name>Maven Central Repository</name>
            <url>https://repo.maven.apache.org/maven2</url>
        </repository>
    </repositories>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
            <!-- Apache Flink dependencies -->
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-api-java-bridge_2.11</artifactId>
                <version>1.14.6</version>
            </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>1.14.6</version>
        </dependency>
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-sql-connector-mysql-cdc</artifactId>
            <version>2.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <version>1.14.6</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.25</version>
        </dependency>
    </dependencies>

</project>

MySQL 数据库

CREATE TABLE `login_detail` (
  `id` int NOT NULL AUTO_INCREMENT,
  `username` varchar(255) DEFAULT NULL,
  `password` varchar(255) DEFAULT NULL,
  `time` varchar(255) DEFAULT NULL,
  `status` int DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=127 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci 

CREATE TABLE `err_login` (
  `id` int NOT NULL AUTO_INCREMENT,
  `username` varchar(255) DEFAULT NULL,
  `status` int DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=74 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

1. 数据模型定义

首先，我们定义了一个简单的 User case class，用于表示从 MySQL 中读取的用户数据。

case class User(id: Int, username: String, password: String, time: String, status: Int)

2.自定义 MySQL 数据源

我们实现了一个自定义的 RichSourceFunction，从 MySQL 数据库中读取数据。该函数会不断地查询数据库，并将新数据发送到 Flink 流中。

class MySQLInsertSource(jdbcUrl: String, username: String, password: String, tableName: String) extends RichSourceFunction[User] {
  @volatile private var isRunning = true
  private var connection: Connection = _
  private var lastMaxTime: String = _

  override def open(parameters: org.apache.flink.configuration.Configuration): Unit = {
    super.open(parameters)
    connection = DriverManager.getConnection(jdbcUrl, username, password)
    // Initial load
    val statement = connection.createStatement()
    val resultSet = statement.executeQuery(s"SELECT * FROM $tableName")
    while (resultSet.next()) {
      val user = User(
        resultSet.getInt("id"),
        resultSet.getString("username"),
        resultSet.getString("password"),
        resultSet.getString("time"),
        resultSet.getInt("status")
      )
      // Update lastMaxTime
      if (lastMaxTime == null || user.time > lastMaxTime) {
        lastMaxTime = user.time
      }
    }
  }

  override def run(ctx: SourceFunction.SourceContext[User]): Unit = {
    val statement = connection.createStatement()
    while (isRunning) {
      val query = s"SELECT * FROM $tableName WHERE time > '$lastMaxTime'"
      val resultSet = statement.executeQuery(query)
      while (resultSet.next()) {
        val user = User(
          resultSet.getInt("id"),
          resultSet.getString("username"),
          resultSet.getString("password"),
          resultSet.getString("time"),
          resultSet.getInt("status")
        )
        ctx.collect(user)
        // Update lastMaxTime
        if (user.time > lastMaxTime) {
          lastMaxTime = user.time
        }
      }
      Thread.sleep(2000) // sleep for 2 seconds
    }
  }

  override def cancel(): Unit = {
    isRunning = false
    if (connection != null) {
      connection.close()
    }
  }
}

变量声明：

isRunning: 用于控制数据源是否继续运行。

connection: 用于连接 MySQL 数据库的 Connection 对象。

lastMaxTime: 记录上次读取数据的最大时间戳，用于增量查询。

open 方法：在数据源启动时初始化数据库连接并进行初始加载，读取全部数据，更新 lastMaxTime。

run 方法：在数据源运行时不断查询数据库，获取新数据并发送到 Flink 流中。每隔2秒执行一次查询，并更新 lastMaxTime。

cancel 方法：在数据源取消时关闭数据库连接。

3. 时间戳分配器和水位线

为了确保事件按时间顺序处理，我们为数据流分配时间戳并生成水位线。

val userStreamWithTimestamps = userStream
  .assignTimestampsAndWatermarks(
    WatermarkStrategy
      .forBoundedOutOfOrderness[User](Duration.ofSeconds(1))
      .withTimestampAssigner(new SerializableTimestampAssigner[User] {
        override def extractTimestamp(element: User, recordTimestamp: Long): Long = {
          val format = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
          val date = format.parse(element.time)
          date.getTime
        }
      })
  )

WatermarkStrategy：定义了水位线生成策略。forBoundedOutOfOrderness 表示允许事件在1秒的乱序范围内到达。

SerializableTimestampAssigner：定义了时间戳提取器，从 User 对象的 time 字段提取时间戳。

4. 数据过滤和窗口处理

我们过滤出 status 为 0 的记录，并对这些记录进行2秒的窗口处理。

val filteredStream = userStreamWithTimestamps.filter(_.status == 0)

val windowedStream = filteredStream
  .keyBy(_.username)
  .timeWindow(Time.seconds(2))
  .process(new WriteToDatabaseFunction(jdbcUrl, username, password))

过滤：filter 操作保留 status 为 0 的记录。（0为登陆失败）

窗口处理：对每个 username 进行2秒的时间窗口处理，并使用自定义的 WriteToDatabaseFunction 进行处理。

5. 窗口处理函数

我们实现了一个 ProcessWindowFunction，在窗口结束时将获取到的异常登陆用户写入 MySQL 数据库。

class WriteToDatabaseFunction(url: String, username: String, password: String) extends ProcessWindowFunction[User, String, String, TimeWindow] {
  val insertSql = "INSERT INTO err_login (username, status) VALUES (?, ?)"

  override def process(key: String, context: Context, elements: Iterable[User], out: Collector[String]): Unit = {
    val allStatusOne = elements.forall(_.status == 0)
    if (allStatusOne) {
      out.collect(s"Username: $key had status 1 for 2 seconds")
      val connection = DriverManager.getConnection(url, username, password)
      val preparedStatement = connection.prepareStatement(insertSql)
      try {
        for (user <- elements) {
          preparedStatement.setString(1, user.username)
          preparedStatement.setInt(2, user.status)
          preparedStatement.addBatch()
        }
        preparedStatement.executeBatch()
      } finally {
        preparedStatement.close()
        connection.close()
      }
    }
  }
}

变量声明：insertSql 为插入错误登录记录的 SQL 语句。

process 方法：

检查窗口内的所有记录 status 是否都为 0。

如果是，打印日志并将记录写入 err_login 表中。

使用批量插入提高效率。

6. 主函数

最后，我们将所有部分组装在一起，并执行 Flink 作业。

object FlinkMySQLExample {
  val jdbcUrl = "jdbc:mysql://localhost:3306/big_data"
  val username = "root"
  val password = "12345678"
  val tableName = "login_detail"

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val mySQLSource = new MySQLInsertSource(jdbcUrl, username, password, tableName)
    val userStream = env.addSource(mySQLSource)

    userStream.print()

    val userStreamWithTimestamps = userStream
      .assignTimestampsAndWatermarks(
        WatermarkStrategy
          .forBoundedOutOfOrderness[User](Duration.ofSeconds(1))
          .withTimestampAssigner(new SerializableTimestampAssigner[User] {
            override def extractTimestamp(element: User, recordTimestamp: Long): Long = {
              val format = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
              val date = format.parse(element.time)
              date.getTime
            }
          })
      )

    val filteredStream = userStreamWithTimestamps.filter(_.status == 0)

    val windowedStream = filteredStream
      .keyBy(_.username)
      .timeWindow(Time.seconds(2))
      .process(new WriteToDatabaseFunction(jdbcUrl, username, password))

    windowedStream.print()

    env.execute("Flink MySQL Example")
  }
}

主函数：

获取 Flink 的执行环境。

添加自定义数据源 MySQLInsertSource，从 MySQL 数据库中读取数据。

将数据流赋予时间戳和水位线。

过滤出 status 为 0 的记录。

对过滤后的记录进行2秒的窗口处理，并将结果写入 MySQL 数据库。

执行 Flink 作业。

7.总结

这段代码展示了如何使用 Apache Flink 处理实时数据流，并与 MySQL 数据库进行交互。通过自定义数据源、时间戳和水位线分配、窗口处理和自定义窗口函数，我们可以构建强大的流处理应用程序。

Flink实现实时异常登陆监控（两秒内多次登陆失败进行异常行为标记）

Flink实现异常登陆监控（两秒内多次登陆失败进行异常行为标记）

项目概述

依赖环境

1. 数据模型定义

2.自定义 MySQL 数据源

3. 时间戳分配器和水位线

4. 数据过滤和窗口处理

5. 窗口处理函数

6. 主函数

7.总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Flink实现实时异常登陆监控（两秒内多次登陆失败进行异常行为标记）

Flink实现异常登陆监控（两秒内多次登陆失败进行异常行为标记）

项目概述

依赖环境

1. 数据模型定义

2.自定义 MySQL 数据源

3. 时间戳分配器和水位线

4. 数据过滤和窗口处理

5. 窗口处理函数

6. 主函数

7.总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景