### 使用 Flink JDBC 输入格式创建 DataStreamSource 为了通过 JDBC 输入格式创建 `DataStreamSource<Row>`，可以利用 Apache Flink 提供的 `JdbcInputFormat` 类来实现这一目标。由于 JDBC 连接器和驱动程序并非 Flink 的默认组件[^1]，因此需要手动配置并引入相应的依赖项。 下面是一个完整的 Java 实现示例： ```java import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.types.Row; import org.apache.flink.api.java.io.jdbc.JDBCInputFormat; public class JdbcDataSourceExample { public static void main(String[] args) throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Row> jdbcResult = env.createInput( JDBCInputFormat.buildJDBCInputFormat() .setDrivername("com.mysql.cj.jdbc.Driver") // 设置数据库驱动名称 .setDBUrl("jdbc:mysql://localhost:3306/testdb") // 数据库连接URL .setQuery("SELECT id, name FROM users LIMIT 5") // SQL查询语句 .setUsername("root") // 用户名 .setPassword("password") // 密码 .finish()); // 打印结果集中的每一行记录 jdbcResult.print(); // 如果是在流环境中，则应使用 StreamExecutionEnvironment 并调用 execute 方法启动作业 } } ``` 上述代码展示了如何定义一个基于批处理环境 (`ExecutionEnvironment`) 来读取来自 MySQL 数据库的数据，并将其作为 Row 对象返回给应用程序。对于实时数据流场景下构建 `DataStreamSource<Row>`，则需采用类似的逻辑但是要切换至 `StreamExecutionEnvironment` 和对应的 API 函数。 需要注意的是，在实际应用中应当替换掉这里的硬编码值（比如主机地址、端口号、用户名密码等），转而从安全的地方获取这些敏感信息；同时也要确保所使用的 JDBC 驱动已经被正确添加到了项目的依赖列表里[^2]。