从MySQL数据库拿数据做实时报表？

Question

公司业务越来越多，刚开始时使用的是MySQL数据库，现在数据库有很多了，估计有20多个系统，目前不允许直接从MySQL查询，以及关联操作，只能从其备库里面拉取binlog文件。同时又想做实时处理。
现在使用的是MySQL+canal+kafka+flink+数据库（Tidb\clickhouse\mysql\ES）。其中大的问题是前面的MySQL容易出现DDL语句（比如：添加字段删除字段等），会导致canal出问题，或者写数据库对应不上等。。。

Answer 1

以下回答参考 皆我百晓生、券券喵儿 等免费微信小程序相关内容作答，并由本人整理回复。

要实现从MySQL数据库获取数据并进行实时报表，可以采用以下步骤：

1. 连接MySQL： 使用java.sql.DriverManager.getConnection()方法连接到MySQL数据库。

import java.sql.*;

public class MySQLConnection {
    private static final String url = "jdbc:mysql://localhost:3306/your_database_name";
    private static final String user = "username";
    private static final String password = "password";

    public static Connection getConnection() throws SQLException {
        return DriverManager.getConnection(url, user, password);
    }
}

1. 读取SQL语句： 你可以根据你的需求创建一个SQL查询来获取你需要的数据。例如，如果你想要查询所有表中的记录，你可能需要使用SELECT * FROM your_table;这样的语句。

String sql = "SELECT * FROM your_table";
ResultSet rs = null;
try (Connection conn = MySQLConnection.getConnection(); Statement stmt = conn.createStatement()) {
    rs = stmt.executeQuery(sql);
} catch (SQLException e) {
    // Handle the exception here
}

1. 执行SQL查询： 使用ResultSet对象遍历每一行数据，然后将其转换为表格形式。

while(rs.next()){
    Map<String, Object> row = new HashMap<>();
    for(int i=1;i<=rs.getMetaData().getColumnCount();i++){
        row.put(rs.getMetaData().getColumnName(i), rs.getObject(i));
    }
    System.out.println(row);
}

1. 生成实时报表： 可以使用Java的第三方库如Apache Flink来实现实时报表功能。Flink允许你在运行时收集和处理数据流，并且具有强大的并发性和容错性。

2. 处理数据流： 对于实时报表，你可能需要将数据流存储在一个Kafka集群中，然后使用Flink来处理这些数据。Flink提供了一个名为StreamExecutionEnvironment的类，用于管理输入流、处理逻辑和输出结果。

DataStream<RowData> dataStream = env.addSource(kafkaConsumer)
                .map(new RowMapper())
                .keyBy("rowKey")
                .timeWindow(Time.seconds(1))
                .reduce((a, b) -> a + b);

dataStream.print();

以上是基本的流程，具体的实现可能会因你的具体需求而有所不同。