关于在Kafka Connect中使用ExtractTopic转换自定义Elasticsearch Sink索引名的可行性咨询

当然可以实现！你当前用的ExtractTopic转换只能提取单个字段值作为索引名，要实现多个字段拼接的需求，这里有两种实用方案供你选择：

方案一：用KSQL预处理消息（无需编写自定义代码） #

在消息进入Kafka Connect之前，先用KSQL创建数据流，把empname和empid拼接成一个新字段，再让Elasticsearch Sink消费这个处理后的流：

1. 首先创建对应原始数据的KSQL流：

CREATE STREAM emp_source_stream (empid STRING, empname STRING) WITH (
    KAFKA_TOPIC='你的原始Topic名称',
    VALUE_FORMAT='JSON'
);

2. 生成包含拼接字段的新流：

CREATE STREAM emp_index_ready_stream WITH (KAFKA_TOPIC='emp-with-index-field') AS
SELECT *, CONCAT(empname, empid) AS index_name
FROM emp_source_stream;

3. 修改Kafka Connect的Elasticsearch Sink配置，让它消费新生成的Topic，并用ExtractTopic提取拼接好的index_name字段作为索引名：

transforms=IndexName
transforms.IndexName.type=io.confluent.connect.transforms.ExtractTopic$Value
transforms.IndexName.field=index_name
transforms.IndexName.skip.missing.or.null=true

这样配置后，生成的索引名就会是test100这类拼接后的名称了。

方案二：自定义Single Message Transform（SMT）（适合有开发能力的场景） #

如果不想引入KSQL组件，可以自己编写一个自定义SMT来实现字段拼接逻辑：

1. 实现org.apache.kafka.connect.transforms.Transformation接口，在apply方法里读取消息中的empid和empname字段，拼接后将值设置为记录的Topic属性（因为Elasticsearch Sink默认会用Topic名作为索引名，或者你也可以通过topic.index.map做映射，但直接修改Topic属性更直接）。

2. 把自定义SMT的jar包放到Kafka Connect的plugin.path指定目录下，让Connect能加载这个插件。

3. 修改Connect配置，使用你的自定义SMT：

transforms=IndexName
transforms.IndexName.type=com.yourcompany.connect.transforms.ConcatFieldsToTopic$Value
transforms.IndexName.target-fields=empname,empid
transforms.IndexName.separator=''  # 这里设置为空，让字段值直接拼接，也可以指定分隔符比如"-"

这种方案灵活性更高，适合需要复杂字段处理的场景，但需要具备基础的Java开发能力。

小提示 #

• 用KSQL方案的话，要确保KSQL集群和Kafka Connect集群能正常通信，并且Topic的权限配置正确。
• 自定义SMT要注意和你使用的Kafka Connect版本兼容，避免出现兼容性问题。

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