Confluent Cloud Elasticsearch连接器如何筛选丢弃/保留指定字段

Confluent Cloud Elasticsearch连接器本身没有专属的批量字段过滤配置，这类字段裁剪需求直接用Kafka Connect框架层通用的单消息转换（SMT）就能实现，不需要二次开发连接器，具体实现分两种场景：

这是成本最低的实现方式，不需要额外部署组件，直接在连接器配置里加转换规则即可。你之前在连接器专属配置里没找到相关选项，是因为这个能力是Kafka Connect框架提供的通用能力，不属于ES连接器的专属配置项。

白名单模式：仅保留指定字段，其余全丢弃 #

适合明确知道需要同步到ES的字段列表的场景，配置示例如下：

# 声明转换器别名
transforms=fieldFilter
# 指定转换器类型，$Value表示处理消息的value部分，要处理key就换成$Key
transforms.fieldFilter.type=org.apache.kafka.connect.transforms.ReplaceField$Value
# 配置要保留的字段列表，用逗号分隔
transforms.fieldFilter.whitelist=user_id,trade_amount,pay_time,order_status

黑名单模式：仅丢弃指定字段，其余全保留 #

适合只需要剔除少数冗余字段、大部分字段都要同步的场景，只需要把上面配置里的whitelist换成blacklist即可：

transforms=fieldFilter
transforms.fieldFilter.type=org.apache.kafka.connect.transforms.ReplaceField$Value
# 配置要丢弃的字段列表，用逗号分隔
transforms.fieldFilter.blacklist=_internal_track_id,tmp_marker,raw_unparsed_content

嵌套字段处理 #

如果你的事件里有嵌套结构，需要过滤嵌套路径下的字段，直接用Confluent Cloud内置的增强版ReplaceField即可，支持点分隔的路径写法，比如要丢弃user.device.fingerprint、extra.temp_tag两个嵌套字段，配置类名换成Confluent增强版即可：

transforms=fieldFilter
transforms.fieldFilter.type=io.confluent.connect.transforms.ReplaceField$Value
transforms.fieldFilter.blacklist=user.device.fingerprint,extra.temp_tag

如果你的字段过滤规则比较复杂，比如要按字段值动态判断是否保留、按前缀/后缀批量匹配字段、过滤规则需要动态更新，SMT满足不了需求的话，可以在数据写入ES连接器消费的源Topic之前，加一层KStreams预处理：

• 写一个轻量KStreams任务消费原始业务Topic
• 在mapValues步骤里完成字段裁剪、规则过滤逻辑
• 处理后的消息输出到专用的ES同步Topic，让ES连接器消费这个Topic即可

这个方案灵活度最高，可以覆盖任意复杂的字段处理需求。

• 配置转换器类名时不要漏写$Value/$Key后缀，否则会抛出类找不到的报错
• 字段匹配是大小写敏感的，配置的字段名必须和事件实际字段名、Schema里的字段名完全一致
• 如果配置了多个SMT，注意调整SMT的执行顺序，字段过滤尽量放在靠后的位置，避免被前面的转换步骤改了字段名导致匹配失败

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