Flink的JDBC连接器sink可以用于删除数据，通过设置DELETE语句和WHERE条件来实现。

Flink中使用JDBC连接器sink删除数据

10年的桐庐网站建设经验，针对设计、前端、开发、售后、文案、推广等六对一服务，响应快，48小时及时工作处理。全网营销推广的优势是能够根据用户设备显示端的尺寸不同，自动调整桐庐建站的显示方式，使网站能够适用不同显示终端，在浏览器中调整网站的宽度，无论在任何一种浏览器上浏览网站，都能展现优雅布局与设计，从而大程度地提升浏览体验。创新互联建站从事“桐庐网站设计”,“桐庐网站推广”以来，每个客户项目都认真落实执行。

使用JDBC连接器sink删除数据

在Flink中，可以使用JDBC连接器的sink来删除数据，具体步骤如下：

1、引入依赖：

```xml

org.apache.flink

flinkconnectorjdbc_2.11

${flink.version}

```

2、创建JDBC连接参数：

```java

Map jdbcOptions = new HashMap<>();

jdbcOptions.put("url", "jdbc:mysql://localhost:3306/mydatabase");

jdbcOptions.put("table", "mytable");

jdbcOptions.put("user", "username");

jdbcOptions.put("password", "password");

```

3、创建JDBC Sink：

```java

JdbcSink sink = JdbcSink.sink(

"INSERT INTO mytable (column1, column2) VALUES (?, ?)",

(ps, t) > {

ps.setString(1, t.getField(0));

ps.setString(2, t.getField(1));

},

jdbcOptions,

new JdbcExecutionOptions.Builder().build()

);

```

4、将数据写入JDBC Sink：

```java

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

FlinkJdbcConnectionOptions options = new FlinkJdbcConnectionOptions.Builder()

.withUrl("jdbc:mysql://localhost:3306/mydatabase")

.withDriverName("com.mysql.jdbc.Driver")

.build();

FlinkJdbcTableEnvironment tableEnv = StreamTableEnvironment.create(env, options);

tableEnv.executeSql("DELETE FROM mytable"); // 删除表中的数据

```

相关问题与解答

问题1：如何在Flink中使用JDBC连接器sink更新数据？

答案：在Flink中使用JDBC连接器sink更新数据，可以按照以下步骤进行操作：

1、创建JDBC连接参数；

2、创建JDBC Sink，并指定更新语句和更新逻辑；

3、将数据写入JDBC Sink。

问题2：如何设置JDBC连接器sink的事务支持？

答案：要设置JDBC连接器sink的事务支持，可以在创建JDBC Sink时添加TransactionConfig配置，示例如下：

TransactionConfig transactionConfig = new TransactionConfig(true, 2); // true表示开启事务支持，2表示事务隔离级别为READ_COMMITTED
JdbcSink sink = JdbcSink.sink(..., ..., jdbcOptions, transactionConfig, new JdbcExecutionOptions.Builder().build());

新闻名称：Flink有没有人用过JDBC连接器sink的时候删除数据？
文章网址：http://www.jxjierui.cn/article/cosehii.html