我正在尝试使用elasticsearch sink连接器将kafka主题中的记录加载到elasticsearch,但是我正在努力按照我想要的方式构造文档id。我希望写入elasticsearch的文档id是由两个值组成的,两个值之间用下划线隔开,这两个值与我的Kafka主题的消息是一致的。
例如:
我的Kafka主题值具有以下avro模式:
{
"type": "record",
"name": "SampleValue",
"namespace": "com.abc.test",
"fields": [
{
"name": "value1",
"type": [
"null",
{
"type": "int",
"java-class": "java.lang.Integer"
}
],
"default": null
},
{
"name": "value2",
"type": [
"null",
{
"type": "int",
"java-class": "java.lang.Integer"
}
],
"default": null
},
{
"name": "otherValue",
"type": [
"null",
{
"type": "int",
"java-class": "java.lang.Integer"
}
],
"default": null
}
]
}
我希望写入elasticsearch的文档id是由下划线分隔的值1和值2的组合值。如果avro中的给定值
{"value1": {"int": 123}, "value2": {"int": 456}, "value3": {"int": 0}}
然后我希望elasticsearch的文档id 123_456
.
我想不出在连接器配置中链接转换的正确方法来创建一个键,该键由两个由下划线分隔的值组成。
1条答案
按热度按时间gstyhher1#
我不认为有一个单一的信息转换开箱即用,将做你想要的。
您可以使用transformapi编写自己的,也可以使用流处理器,如kafka streams或ksqldb。