基于Flink MySQL CDC技术实现交易告警

前言

CDC 的全称是 Change Data Capture，是一种用于捕获数据库变更数据的技术。例如 MySQL 对数据的所有变更都会写入到 binlog，CDC 就可以通过监听 binlog 文件来实现对 MySQL 数据变更的捕获，然后做进一步的处理。

Flink CDC 将CDC技术和 Flink 流计算整合到一起，把CDC捕获到的数据变更作为 Flink数据源，以实现对数据变更的流式处理。通过 Flink CDC 可以轻松实现以下功能：

数据同步将一个数据库中的数据变化实时同步到另一个数据库或数据存储中，实现数据的实时备份和迁移
实时数据分析捕获数据库的变更数据，并将其作为实时流数据源输入到 Flink 进行实时分析和处理，进行实时报表生成、实时监控、实时推荐等应用
数据集成和 ETL 在数据集成和 ETL过程中，使用 Flink CDC 可以实现对源数据库的实时数据抽取，然后进行数据转换和加载到目标系统中

本文就来实现一个简单的 Flink 作业，通过 Flink CDC 技术来监控 MySQL 中的用户交易记录，针对频繁交易和大额交易进行风控告警。

需求描述

用户交易记录存储在 MySQL 的 user_trade 表中，编写一个 Flink 作业实现对用户交易记录的监听，针对每个用户在一分钟内，若交易次数超过十次，或者交易金额超过一万元的，生成一条交易告警记录并写入 user_trade_alert 表，由业务系统触发告警操作。

需求实现

前期准备

执行DDL语句，完成表的创建

CREATE TABLE user_trade
(id         BIGINT(20) NOT NULL AUTO_INCREMENT PRIMARY KEY,user_id    BIGINT(20) NOT NULL,amount     BIGINT(20) NOT NULL,trade_time DATETIME   NOT NULL
) COMMENT '用户交易';
CREATE TABLE user_trade_alert
(id           BIGINT(20)    NOT NULL AUTO_INCREMENT PRIMARY KEY,user_id      BIGINT(20)    NOT NULL,alert_reason VARCHAR(1024) NOT NULL
) COMMENT '用户交易告警';

引入Maven依赖，mysql-connector-java 是基础，因为要读写MySQL数据库；flink-connector-jdbc 是 Flink 提供的JDBC 连接器，用于 Flink 读写数据库；flink-connector-mysql-cdc 是 Flink 提供的针对 MySQL 的 CDC 实现。

<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.31</version>
</dependency>
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc</artifactId><version>3.2.0-1.19</version>
</dependency>
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>3.2.0</version>
</dependency>

Flink作业编写

1、先对数据实体建模。UserTrade 对应用户交易记录，UserTradeStat 对应用户交易的统计，UserTradeAlert 对应用户交易的告警。

@Data
public static class UserTrade {private Long id;@JSONField(name = "user_id")private Long userId;private Long amount;@JSONField(name = "trade_time")private Long tradeTime;
}@Data
@NoArgsConstructor
@AllArgsConstructor
public static class UserTradeAlert {private Long userId;private String alertReason;
}@Data
@NoArgsConstructor
@AllArgsConstructor
public static class UserTradeStat {private Long userId;private long totalAmount;private long total;
}

2、构建 MySqlSource，Flink cdc 的核心数据源，它会连接到 MySQL 主库并消费 binlog 日志来获取数据变更记录，然后转发给下游算子处理。

private static MySqlSource<UserTrade> ofMySqlSource(String database, String table) {return MySqlSource.<UserTrade>builder().hostname(DB_HOST).port(DB_PORT).databaseList(database).tableList(database + "." + table).username(DB_USERNAME).password(DB_PASSWORD).deserializer(new UserTradeDeserialization()).build();
}

Flink CDC 获取到的数据被封装成org.apache.kafka.connect.source.SourceRecord 类，我们要自定义反序列化器，将获取到的日志记录转化成 UserTrade 实体对象。

public static class UserTradeDeserialization implements DebeziumDeserializationSchema<UserTrade> {transient JsonConverter jsonConverter;@Overridepublic void deserialize(SourceRecord record, Collector<UserTrade> collector) throws Exception {// 交易数据，不考虑[删改]的场景byte[] bytes = getJsonConverter().fromConnectData(record.topic(), record.valueSchema(), record.value());UserTrade userTrade = JSON.parseObject(bytes).getJSONObject("payload").getObject("after", UserTrade.class);// 时区问题userTrade.setTradeTime(userTrade.getTradeTime() - Duration.ofHours(8L).toMillis());collector.collect(userTrade);}@Overridepublic TypeInformation<UserTrade> getProducedType() {return TypeInformation.of(UserTrade.class);}JsonConverter getJsonConverter() {if (jsonConverter == null) {this.jsonConverter = new JsonConverter();HashMap<String, Object> configs = new HashMap(2);configs.put("converter.type", ConverterType.VALUE.getName());configs.put("schemas.enable", true);this.jsonConverter.configure(configs);}return jsonConverter;}
}

3、因为是统计每分钟内用户的交易次数和交易额度，必然要用到窗口计算，所以要编写窗口处理函数。UserTradeAggregateFunction 是对窗口内数据的聚合处理，UserTradeWindowFunction 是对窗口计算结果的处理，如果满足告警规则，则会生成一个 UserTradeAlert 对象交给下游算子处理。

public static class UserTradeAggregateFunction implements AggregateFunction<UserTrade, UserTradeStat, UserTradeStat> {@Overridepublic UserTradeStat createAccumulator() {return new UserTradeStat();}@Overridepublic UserTradeStat add(UserTrade value, UserTradeStat accumulator) {System.err.println("add:" + value);accumulator.setUserId(value.getUserId());accumulator.setTotalAmount(accumulator.getTotalAmount() + value.amount);accumulator.setTotal(accumulator.getTotal() + 1);return accumulator;}@Overridepublic UserTradeStat getResult(UserTradeStat accumulator) {return accumulator;}@Overridepublic UserTradeStat merge(UserTradeStat a, UserTradeStat b) {return null;}
}public static class UserTradeWindowFunction implements WindowFunction<UserTradeStat, UserTradeAlert, Long, TimeWindow> {@Overridepublic void apply(Long key, TimeWindow timeWindow, Iterable<UserTradeStat> iterable, Collector<UserTradeAlert> collector) throws Exception {System.err.println("win:" + timeWindow.getStart() + "," + timeWindow.getEnd());iterable.forEach(stat -> {if (stat.getTotal() > 10 || stat.getTotalAmount() > 1000000) {LocalDateTime startTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timeWindow.getStart()), ZoneId.systemDefault());LocalDateTime endTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timeWindow.getEnd()), ZoneId.systemDefault());String alertReason = "用户: " + key + " 在[" + startTime + "," + endTime + "]期间产生交易:"+ stat.getTotal() + "笔,交易金额:" + stat.getTotalAmount() + ",触发告警规则.";collector.collect(new UserTradeAlert(key, alertReason));}});}
}

4、窗口算子生成的 UserTradeAlert 对象会交给 Sink 算子写进数据库里，所以最后还差一个 SinkFunction，因为是写进MySQL，所以这里构建一个 JdbcSink。

private static SinkFunction<UserTradeAlert> ofMysqlSink() {return JdbcSink.<UserTradeAlert>sink("INSERT INTO user_trade_alert (user_id,alert_reason) VALUES (?,?)",(ps, value) -> {ps.setLong(1, value.getUserId());ps.setString(2, value.getAlertReason());}, JdbcExecutionOptions.builder().withBatchIntervalMs(100).withMaxRetries(0).build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://ip:port/flink_db?useSSL=false").withUsername(DB_USERNAME).withPassword(DB_PASSWORD).withDriverName("com.mysql.cj.jdbc.Driver").build());
}

5、所有组件都写完了，最后就是启动 Flink 执行环境，把整个作业串起来。

private static final String DB_HOST = "your_ip";
private static final int DB_PORT = 3306;
private static final String DB_USERNAME = "admin";
private static final String DB_PASSWORD = "xxx";public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();environment.fromSource(ofMySqlSource("flink_db", "user_trade"), WatermarkStrategy.<UserTrade>forMonotonousTimestamps().withTimestampAssigner((value, time) -> value.getTradeTime()), "mysql-cdc").setParallelism(1).keyBy(UserTrade::getUserId).window(TumblingEventTimeWindows.of(Duration.ofSeconds(60L))).aggregate(new UserTradeAggregateFunction(), new UserTradeWindowFunction()).addSink(ofMysqlSink());environment.execute();
}

功能验证

提交并运行Flink作业，往MySQL user_trade 表写入一些用户交易记录，当这些交易记录触发告警规则时，Flink 作业就会往 user_trade_alert 表生成如下告警记录示例：

id	user_id	alert_reason
1	1	用户: 1 在[2024-09-13T15:29:10,2024-09-13T15:29:15]期间产生交易:1笔,交易金额:9999999,触发告警规则.
3	2	用户: 2 在[2024-09-13T15:30:25,2024-09-13T15:30:30]期间产生交易:11笔,交易金额:21,触发告警规则.