如何在Kafka Elasticsearch Sink Connector中组合两个字段设置文档ID

完全可以通过现有Kafka Connect SMT（单消息转换）实现将两个字段组合为文档ID的需求，无需额外开发自定义逻辑，可参考以下方案执行：

• 如果你使用的是2.6及以上版本的Kafka Connect，可直接通过ValueToKey + Concat 两个SMT组合完成配置，参考配置片段如下：
  1. 先将目标字段提取到消息Key中

  transforms=extractKeyFields
  transforms.extractKeyFields.type=org.apache.kafka.connect.transforms.ValueToKey
  transforms.extractKeyFields.fields=Event_start_time,allbeat.heartbeat.target
  

  2. 拼接两个字段为单一字符串作为最终ID，可自定义分隔符（如下划线、中划线等）

  transforms=concatId
  transforms.concatId.type=io.confluent.connect.transforms.Concat$Key
  transforms.concatId.fields=Event_start_time,allbeat.heartbeat.target
  transforms.concatId.delimiter=_
  transforms.concatId.alias=_id
  

• 如果使用的是低版本Kafka Connect不支持直接读取嵌套字段，可先通过ExtractNestedField SMT将allbeat.heartbeat.target提取为顶层字段，再执行上述提取和拼接操作即可。
• 若你对接的是Elasticsearch sink连接器，也可以跳过拼接SMT步骤，直接在连接器的document.id配置项中指定两个字段的组合规则，连接器会自动生成对应ID。

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