基于Kafka JSON数据，如何用Apache Flink实现JDBC Sink的增删操作？

一、如何通过Flink从JDBC Sink中删除记录 #

Flink的JDBC Sink默认主打插入和更新操作，但要实现删除，有两种实用的思路：

1. 自定义JDBC Statement Builder #

这是最直接的方式——通过实现JdbcStatementBuilder接口，根据数据里的操作标记动态生成DELETE语句。举个实际例子：
假设你的数据里有个operation字段，值为"delete"时执行删除，否则执行插入：

JdbcSink.sink(
    "", // 这里不需要固定SQL，由builder动态生成
    (statement, row) -> {
        String op = row.getFieldAs("operation");
        Integer id = row.getFieldAs("id");
        if ("delete".equals(op)) {
            // 按主键构造删除语句
            statement.executeUpdate("DELETE FROM your_table WHERE id = ?");
            statement.setInt(1, id);
        } else {
            // 常规插入逻辑
            statement.executeUpdate("INSERT INTO your_table (id, name) VALUES (?, ?)");
            statement.setInt(1, id);
            statement.setString(2, row.getFieldAs("name"));
        }
    },
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://localhost:3306/your_db")
        .withUsername("db_user")
        .withPassword("db_pass")
        .build()
);

这种方式需要确保数据里有能精准定位删除记录的标识（比如主键），同时要做好事务和异常处理，避免数据不一致。

2. 用Upsert模式结合「墓碑记录」 #

如果你的场景是基于主键的变更，也可以开启Upsert模式的JDBC Sink，通过发送墓碑记录（即仅含主键、其他字段为null或带删除标记的记录）触发删除。不过这种方式需要在Upsert SQL里做额外判断，比如：

String upsertSql = "INSERT INTO your_table (id, name, is_delete) VALUES (?, ?, ?) " +
                   "ON DUPLICATE KEY UPDATE " +
                   "name = VALUES(name), is_delete = VALUES(is_delete); " +
                   "DELETE FROM your_table WHERE is_delete = 1";

这种方式相对绕一些，更推荐第一种自定义Statement Builder的方案，逻辑更直观。

二、从Kafka读取JSON数据，根据字段值执行JDBC Sink的插入或删除 #

这个场景是第一个问题的延伸，整体分两步走：

1. 解析Kafka中的JSON数据 #

先用Flink Kafka Consumer读取JSON数据，再解析成Flink能处理的Row或POJO。比如用Jackson解析：

DataStream<Row> kafkaStream = env.fromSource(
    KafkaSource.<String>builder()
        .setBootstrapServers("kafka:9092")
        .setTopics("your_topic")
        .setGroupId("flink-jdbc-group")
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build(),
    WatermarkStrategy.noWatermarks(),
    "Kafka JSON Source"
)
.map(jsonStr -> {
    ObjectMapper mapper = new ObjectMapper();
    JsonNode node = mapper.readTree(jsonStr);
    // 假设JSON里有id、name、op（操作类型：insert/delete）字段
    return Row.of(
        node.get("id").asInt(),
        node.get("name").asText(),
        node.get("op").asText()
    );
});

2. 根据操作字段路由JDBC操作 #

接着用自定义Statement Builder，根据op字段的值判断执行INSERT还是DELETE：

kafkaStream.addSink(JdbcSink.sink(
    "",
    (statement, row) -> {
        String op = row.getFieldAs(2);
        Integer id = row.getFieldAs(0);
        String name = row.getFieldAs(1);
        
        switch(op) {
            case "insert":
                statement.executeUpdate("INSERT INTO your_table (id, name) VALUES (?, ?)");
                statement.setInt(1, id);
                statement.setString(2, name);
                break;
            case "delete":
                statement.executeUpdate("DELETE FROM your_table WHERE id = ?");
                statement.setInt(1, id);
                break;
            case "update":
                // 可选：扩展支持更新操作
                statement.executeUpdate("UPDATE your_table SET name = ? WHERE id = ?");
                statement.setString(1, name);
                statement.setInt(2, id);
                break;
            default:
                // 处理未知操作，比如跳过或抛出异常
                throw new IllegalArgumentException("Unsupported operation type: " + op);
        }
    },
    JdbcExecutionOptions.builder()
        .withBatchSize(1000) // 开启批量提交提升性能
        .withBatchIntervalMs(5000)
        .withMaxRetries(3)
        .build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://localhost:3306/your_db")
        .withUsername("db_user")
        .withPassword("db_pass")
        .build()
)).name("JDBC Upsert/Delete Sink");

额外注意点 #

• 确保Kafka的JSON数据里有明确的操作标记（比如op字段），或者能通过字段完整性判断操作类型（比如删除操作仅包含主键）。
• 异常处理要到位：比如未知操作类型、数据库连接失败时，可以添加重试机制或死信队列，避免数据丢失。

内容的提问来源于stack exchange，提问作者user432024