如何从flink作业中查找数据库数据

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()
    .setDrivername("com.mysql.jdbc.Driver")
    .setDBUrl("jdbc:mysql://localhost:3306/db_name")
    .setUsername("username")
    .setPassword("password")
    .setQuery("SELECT * FROM table_name")
    .setRowTypeInfo(new RowTypeInfo(...))
    .finish();

DataStream<Row> dataStream = env.createInput(jdbcInputFormat);

public class JDBCTableSource implements TableSource<Row> {
    // 实现TableSource接口的方法，包括getTableSchema、getBoundedness、getDataStream等

    @Override
    public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
        JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()
            .setDrivername("com.mysql.jdbc.Driver")
            .setDBUrl("jdbc:mysql://localhost:3306/db_name")
            .setUsername("username")
            .setPassword("password")
            .setQuery("SELECT * FROM table_name")
            .setRowTypeInfo(new RowTypeInfo(...))
            .finish();

        return execEnv.createInput(jdbcInputFormat);
    }
}

// 使用自定义的TableSource
TableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
tableEnv.registerTableSource("myTable", new JDBCTableSource());

Table table = tableEnv.scan("myTable");