能否通过Kafka Sink Connector结合SMT实现单条消息多Elasticsearch索引投递？

1. 用SMT处理复杂JSON并路由到不同ES索引 #

针对复杂JSON消息，可通过内置SMT组合或自定义SMT完成字段提取、计算转换，再结合ES Sink的动态路由能力投递到目标索引：

• 字段提取与转换：使用内置SMT如ExtractField$Value提取嵌套字段、ReplaceField调整字段名，或编写自定义SMT实现复杂计算（比如基于多个嵌套字段生成新的业务标识）。
• 动态索引路由：
  • 方式一：用RegexRouter SMT，基于消息字段值替换索引名。例如从JSON中提取user.type字段，将索引设置为user_${type}：

    transforms=extractType,routeIndex
    transforms.extractType.type=org.apache.kafka.connect.transforms.ExtractField$Value
    transforms.extractType.field=user.type
    transforms.routeIndex.type=org.apache.kafka.connect.transforms.RegexRouter
    transforms.routeIndex.regex=(.*)
    transforms.routeIndex.replacement=user_$1
    

  • 方式二：自定义SMT为消息添加es.index.name头信息，ES Sink会自动读取该头作为目标索引名，这种方式支持更复杂的计算逻辑（比如根据日期+业务类型组合索引名）。

2. 单条消息发送到多个ES索引 #

目前没有原生开箱即用的方案，但可通过以下两种方式实现：

• Kafka Streams拆分复制消息：读取原Topic后，对单条消息生成多条输出，分别发送到不同的Kafka Topic，每个Topic对应一个ES Sink Connector，各自写入目标索引。示例伪代码：

  StreamsBuilder builder = new StreamsBuilder();
  KStream<String, JsonNode> sourceStream = builder.stream("input-topic");
  
  // 根据业务规则拆分消息到不同流
  KStream<String, JsonNode> streamA = sourceStream.filter((k, v) -> v.get("biz_tag").asText().equals("A"));
  KStream<String, JsonNode> streamB = sourceStream.filter((k, v) -> v.get("biz_tag").asText().equals("B"));
  
  // 发送到不同输出Topic
  streamA.to("output-topic-A");
  streamB.to("output-topic-B");
  

  随后为每个输出Topic配置ES Sink，分别指向index-A和index-B。
• 自定义SMT复制消息：编写自定义SMT，将单条输入消息转换为多条ConnectRecord，每条记录设置不同的es.index.name头，ES Sink处理时会将每条记录写入对应索引。这种方式无需额外Topic，直接在Connect链路内完成复制路由。

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