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 数据库进行交互。通过自定义数据源、时间戳和水位线分配、窗口处理和自定义窗口函数，我们可以构建强大的流处理应用程序。

如有遇到问题可以找小编沟通交流哦。另外小编帮忙辅导大课作业，学生毕设等。不限于MapReduce， MySQL, python，java，大数据，模型训练等。 hadoop hdfs yarn spark Django flask flink kafka flume datax sqoop seatunnel echart可视化机器学习等
在这里插入图片描述