数据同步演进：从 MySQL 实时数据同步 Kafka 到 Flink 实时计算

2024-02-10 13:55:32

很多 DBA 同学经常会遇到要从一个数据库实时同步到另一个数据库的问题，同构数据还相对容易，遇上异构数据、表多、字段多的时候，该怎么操作呢？

数据同步演进

数据同步技术演进经历了三个阶段：

定时轮询同步 ：定时轮询数据库，如果有新数据就同步过来。
触发器同步 ：在数据库上配置触发器，当数据发生变化时，触发器就会触发，把变化的数据同步过来。
流复制同步 ：数据库本身支持流复制功能，可以把数据变化实时同步过来。

MySQL 是世界上最流行的关系型数据库之一，它支持触发器同步和流复制同步。Kafka 是一个分布式消息系统，它可以把数据实时同步到多个消费者。Flink 是一个分布式计算引擎，它可以从 Kafka 实时消费数据，并进行各种计算处理。

案例分享：MySQL 实时数据同步到 Kafka，再由 Flink 实时计算

案例背景 ：

一个电商网站需要把 MySQL 数据库中的订单数据实时同步到 Kafka，再由 Flink 实时计算出订单总金额、订单数量等指标，并写入另一个数据库。

解决方案 ：

在 MySQL 数据库上配置触发器，当订单数据发生变化时，触发器就会触发，把变化的数据写入 Kafka。
使用 Flink 从 Kafka 实时消费数据，并进行计算处理。
把计算结果写入另一个数据库。

具体步骤 ：

在 MySQL 数据库上配置触发器：

CREATE TRIGGER order_update_trigger
AFTER UPDATE ON orders
FOR EACH ROW
BEGIN
    INSERT INTO kafka_orders (order_id, order_amount, order_status)
    VALUES (NEW.order_id, NEW.order_amount, NEW.order_status);
END;

使用 Flink 从 Kafka 实时消费数据：

// 创建 Flink 流处理环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// 设置并行度
env.setParallelism(1);

// 创建 Kafka 数据源
KafkaSource<String> source = new KafkaSource<>(
    "kafka_orders",
    new SimpleStringSchema(),
    PropertiesUtil.buildKafkaProps()
);

// 从 Kafka 实时消费数据
DataStream<String> orders = env.addSource(source);

// 计算订单总金额和订单数量
DataStream<Tuple2<Long, Integer>> result = orders
    .map(new MapFunction<String, Tuple2<Long, Integer>>() {
        @Override
        public Tuple2<Long, Integer> map(String value) throws Exception {
            String[] fields = value.split(",");
            Long orderAmount = Long.valueOf(fields[1]);
            return new Tuple2<>(orderAmount, 1);
        }
    })
    .keyBy(0)
    .sum(1);

// 把计算结果写入另一个数据库
result.addSink(new JdbcSink());

// 启动 Flink 作业
env.execute();

把计算结果写入另一个数据库：

public class JdbcSink implements SinkFunction<Tuple2<Long, Integer>> {

    private PreparedStatement ps;
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取数据库连接
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/flink_result", "root", "password");

        // 创建预编译语句
        ps = connection.prepareStatement("INSERT INTO order_summary (order_amount, order_count) VALUES (?, ?)");
    }

    @Override
    public void invoke(Tuple2<Long, Integer> value, Context context) throws Exception {
        // 设置预编译语句参数
        ps.setLong(1, value.f0);
        ps.setInt(2, value.f1);

        // 执行预编译语句
        ps.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        // 关闭预编译语句和数据库连接
        ps.close();
        connection.close();
    }
}