JDBC Connector

该连接器可以向 JDBC 数据库写入数据。

添加下面的依赖以便使用该连接器(同时添加 JDBC 驱动):

  1. <dependency>
  2. <groupId>org.apache.flink</groupId>
  3. <artifactId>flink-connector-jdbc_2.11</artifactId>
  4. <version>1.12.0</version>
  5. </dependency>

注意该连接器目前还 不是 二进制发行版的一部分,如何在集群中运行请参考 这里

已创建的 JDBC Sink 能够保证至少一次的语义。 更有效的精确执行一次可以通过 upsert 语句或幂等更新实现。

用法示例:

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. env
  3. .fromElements(...)
  4. .addSink(JdbcSink.sink(
  5. "insert into books (id, title, author, price, qty) values (?,?,?,?,?)",
  6. (ps, t) -> {
  7. ps.setInt(1, t.id);
  8. ps.setString(2, t.title);
  9. ps.setString(3, t.author);
  10. ps.setDouble(4, t.price);
  11. ps.setInt(5, t.qty);
  12. },
  13. new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
  14. .withUrl(getDbMetadata().getUrl())
  15. .withDriverName(getDbMetadata().getDriverClass())
  16. .build()));
  17. env.execute();

更多细节请查看 API documentation