flink如何实时从mysql获取数据

Flink 可以通过实现自定义的 SourceFunction 来从 MySQL 中获取数据。具体步骤如下：

1.引入相关依赖。需要引入 flink-sql-connector-mysql 依赖，该依赖为 Flink 官方提供的 MySQL 连接器。

2.实现自定义的 SourceFunction。需要实现 SourceFunction 接口的 run() 和 cancel() 方法。在 run() 方法中，可以使用 JDBC 连接 MySQL 数据库，并执行查询语句，将查询结果作为 Flink 的数据源进行处理。在 cancel() 方法中，可以将连接关闭。

3.将自定义的 SourceFunction 添加到 Flink 程序中。可以通过 StreamExecutionEnvironment 的 addSource() 方法将自定义的 SourceFunction 添加到 Flink 程序中。

示例代码如下：

public class MySQLSourceFunction implements SourceFunction<Row> {
    private final String driverClassName = "com.mysql.jdbc.Driver";
    private final String dbUrl = "jdbc:mysql://localhost:3306/test";
    private final String query = "SELECT * FROM my_table";
    private final String username = "root";
    private final String password = "password";
    private Connection connection = null;
    private PreparedStatement statement = null;
    private ResultSet resultSet = null;

    @Override
    public void run(SourceContext<Row> ctx) throws Exception {
        // 加载驱动
        Class.forName(driverClassName);
        // 连接数据库
        connection = DriverManager.getConnection(dbUrl, username, password);
        // 执行查询
        statement = connection.prepareStatement(query);
        resultSet = statement.executeQuery();
        // 处理查询结果
        while (resultSet.next()) {
            Row row = new Row(2);
            row.setField(0, resultSet.getInt("id"));
            row.setField(1, resultSet.getString("name"));
            ctx.collect(row);
        }
    }

    @Override
    public void cancel() {
        // 关闭连接
        try {
            if (resultSet != null) {
                resultSet.close();
            }
            if (statement != null) {
                statement.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

// 将自定义的 SourceFunction 添加到 Flink 程序中
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new MySQLSourceFunction()).print();
env.execute();