elasticsearch - Kafka连接 Elasticsearch ID创建多个字段不起作用

Question

我在问这个问题，因为在原始案例中没有答案:Elastic Kafka Connector, ID Creation .

我也有类似的情况。

通过 kafkaconnect 发送请求时， Elasticsearch 表为单个字段创建记录，而不是为多个字段创建记录。

获取异常 “ key 用作文档ID，不能为空”在 Elasticsearch 中。

我的连接器配置:

{
 "name": "test-connector33",
 "config": {
 "connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
 "tasks.max": "1",
 "topics": "test-connector33",
 "connection.url": "http://localhost:9200",
 "type.name": "aggregator",
 "schema.ignore": "true",
 "topic.schema.ignore": "true",
  "topic.key.ignore": "false",
 "value.converter": "org.apache.kafka.connect.json.JsonConverter",
 "value.converter.schemas.enable": "false", 
 "key.converter": "org.apache.kafka.connect.json.JsonConverter",
 "key.converter.schemas.enable": "false",
 "key.ignore":"false",
 "name": "test-connector33",
"transforms": "InsertKey,extractKey",
"transforms.InsertKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
"transforms.InsertKey.fields":"customerId,city",
"transforms.extractKey.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
"transforms.extractKey.field":"customerId,city"
}}

知道如何解决这个问题吗？

提前致谢!

Answer 1

org.apache.kafka.connect.transforms.ExtractField$Key仅支持单个字段。

假设您的 JSON 对象是 HashMap<String, Object> .您找不到字段 customerId,city ，所以 map.get(field)操作返回 null ，因此将该字段设置为空。

如果您想通过控制台生产者发送 key ，欢迎添加 --property print.key=true作为标志，然后键入键，按 Tab，然后输入值。如果要将数据回显到进程中，还可以设置--property key.separator='|' , 对于竖线以及添加 --property parse.key=true