基于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;

    @Override
    public 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);
    }

    @Override
    public 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> {
    @Override
    public UserTradeStat createAccumulator() {
        return new UserTradeStat();
    }

    @Override
    public 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;
    }

    @Override
    public UserTradeStat getResult(UserTradeStat accumulator) {
        return accumulator;
    }

    @Override
    public UserTradeStat merge(UserTradeStat a, UserTradeStat b) {
        return null;
    }
}

public static class UserTradeWindowFunction implements WindowFunction<UserTradeStat, UserTradeAlert, Long, TimeWindow> {
    @Override
    public 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 表生成如下告警记录示例：

尾巴

Flink MySQL CDC 具有强大的功能特性，为实时数据处理提供了高效可靠的解决方案。它能够实现对 MySQL 数据库的实时数据捕获，确保数据的及时性和准确性。可以快速响应数据库中的数据变化，将变更数据实时传输到 Flink 流处理引擎中进行进一步的分析和处理。