如何移除Debezium为SQL Server更新生成Payload中的'before'键

针对你的场景（Debezium+Kafka+Hudi数据湖，已跳过删除事件，需缩减Payload体积），以下是三种可行方案，按推荐优先级排序：

1. 从Debezium源头禁用before字段输出（最优） #

直接修改SQL Server Debezium连接器的配置，让它在生成更新事件时完全不输出before字段，这是效率最高的方式，不需要额外的中间处理。

在连接器配置文件中添加/修改以下参数：

# 核心配置：更新事件不包含before字段（Debezium 1.4+版本支持）
include.before.update=false
# 你已配置的跳过删除事件的参数（确认保留）
event.processing.failure.handling.mode=skip
# 可选：如果不需要Schema变更事件，关闭进一步缩减体积
include.schema.changes=false

说明：include.before.update=false是专门针对更新事件的参数，只会移除更新事件的before字段，不影响其他事件的结构，且完全不会扁平化Payload。

2. 使用Kafka Connect SMT过滤before字段 #

如果无法修改Debezium连接器配置（比如共享连接器实例），可以用Kafka Connect自带的ReplaceField单消息转换（SMT）来移除before字段，同样不会扁平化Payload。

在Kafka Connect的连接器配置中添加SMT配置：

# 配置SMT
transforms=removeBefore
transforms.removeBefore.type=org.apache.kafka.connect.transforms.ReplaceField$Value
# 方式一：白名单模式，只保留需要的字段（推荐，避免意外删除必要元数据）
transforms.removeBefore.whitelist=after,op,ts_ms,source
# 方式二：黑名单模式，直接移除before字段
# transforms.removeBefore.blacklist=before

说明：ReplaceField$Value指定只处理消息的Value部分，白名单模式可以精准控制保留的字段（比如op记录操作类型，ts_ms记录时间，source记录数据源信息，这些Hudi可能需要），不会改变Payload的层级结构。

3. Kafka Streams自定义处理（适合复杂场景） #

如果需要结合其他业务逻辑（比如同时过滤特定字段、转换格式），可以编写Kafka Streams应用来移除before字段。示例Java代码片段：

StreamsBuilder builder = new StreamsBuilder();
// 从输入主题读取消息，Value解析为JsonNode
KStream<String, JsonNode> inputStream = builder.stream("your-debezium-topic");

// 移除before字段
KStream<String, JsonNode> outputStream = inputStream.mapValues(value -> {
    if (value.has("before")) {
        value.remove("before");
    }
    return value;
});

// 输出到处理后的主题
outputStream.to("hudi-ingest-topic");

// 启动Streams应用
KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig());
streams.start();

说明：这种方式需要开发和维护代码，适合有自定义需求的场景，比如需要对Payload做其他修改时使用。

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