弹性搜索 - Kafka Connect 未将正确的密钥值发布为文档 ID

Question

我正在尝试使用 Kafka connect 将数据从 kafka 主题发布到 Elastic 搜索。 下面是我的配置。

{
"name": "elasticsearch_sink_19",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "k-connect-status",
"connection.url": "http://docker.for.mac.host.internal:9200",
"type.name": "connectstatus",
"behavior.on.malformed.documents": "ignore",
"errors.tolerance": "all",
"errors.log.enable": "true",
"errors.log.include.messages": "true",
"value.converter": "org.apache.kafka.connect.json.JsonConverter", 
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"validate.non.null": "false",
"key.ignore":"true",
"schema.ignore":"true",
"value.converter.schemas.enable": "false"
}
}

"key.ignore"true 表示发布具有错误_id的数据(_id不是预期的(。

来自主题的数据：

格式：JSON

{
"ROWTIME":1591743374742,
"ROWKEY":"status-connector-elasticsearch_sink_31",
"state":"RUNNING",
"trace":null,
"worker_id":"connect:8083",
"generation":2
}

示例弹性搜索输出：

{
"_index" : "k-connect-status",
"_type" : "connectstatus",
"_id" : "k-connect-status+1+17",
"_score" : 1.0,
"_source" : {
"generation" : 11,
"trace" : null,
"state" : "UNASSIGNED",
"worker_id" : "connect:8083"
}
}

预期的弹性搜索输出

{
"_index" : "k-connect-status",
"_type" : "connectstatus",
"_id" : "status-connector-elasticsearch_sink_31",
"_score" : 1.0,
"_source" : {
"generation" : 11,
"trace" : null,
"state" : "UNASSIGNED",
"worker_id" : "connect:8083"
}
}

ROWKEY 作为弹性搜索的_id。至少附加了主题名称的 ROWKEY。

"key.ignore" false 不发布任何数据。

Answer 1

"key.ignore":"true",正在做它应该做的事情。它忽略了 Kafka 消息的键，而是使用 topic+partition+offset 的元组，这是你可以在引用的 Elasticsearch 输出中看到的：

"_id" : "k-connect-status+1+17",

如果要使用 Kafka 消息的键，则需要设置"key.ignore":"false"

查看本教程以了解更多信息并查看按键视频/代码的说明

• 🎥 在这里观看： https://rmoff.dev/kafka-elasticsearch-video
• 👾 在这里试试：https://rmoff.dev/kafka-elasticsearch

Answer 2

默认情况下，Elasticsearch 接收器连接器通过连接以下内容来创建文档_id：主题、分区和偏移量。它发生在你的位置。 您应该在false上设置key.ignore并提供正确的密钥。

如果需要从消息值中提取一些信息并将其放入键中，则应使用适当的转换。我想你可以试试ValueToKey。