为 Flink Job 创建可分页的 JDBC 源代码

Question

为了处理来自数据库的数据，我正在使用flink。我已经用jdbc创建了输入。

val inputFormat = JDBCInputFormat.buildJDBCInputFormat()
.setDrivername(driver)
.setDBUrl(url)
.setUsername(username)
.setPassword(password)
.setQuery("select id, name from users")
.finish()
env.createInput(inputFormat)

问题是此输入正在从表中获取所有数据。由于此表包含大量信息，因此我需要可分页的jdbc源之类的内容。我可以使用任何其他设置吗？

Answer 1

通过将查询指定为参数化查询并提供要绑定到参数的值，可以将查询拆分为多个部分，这些部分独立执行。

以下内容摘自JDBCInputFormat的JavaDoc。

* <p>In order to query the JDBC source in parallel, you need to provide a
* parameterized query template (i.e. a valid {@link PreparedStatement}) and
* a {@link ParameterValuesProvider} which provides binding values for the
* query parameters. E.g.:
*
* <pre><code>
*
* Serializable[][] queryParameters = new String[2][1];
* queryParameters[0] = new String[]{"Kumar"};
* queryParameters[1] = new String[]{"Tan Ah Teck"};
*
* JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()
*              .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
*              .setDBUrl("jdbc:derby:memory:ebookshop")
*              .setQuery("select * from books WHERE author = ?")
*              .setRowTypeInfo(rowTypeInfo)
*              .setParametersProvider(new GenericParameterValuesProvider(queryParameters))
*              .finish();
* </code></pre>

请注意：

• 查询的表应在参数化属性上具有适当的索引。否则，您将在桌子上进行多次完整扫描，这没有帮助。
• 参数应仅涵盖表的所有(必需(数据一次。否则，您可能会错过某些行或查询某些行两次。