kafka-connect -asticsearch: хранение сообщений в формате заданного индекса

Question

Пример:

{"id":"1","firstName":"abc","lastName":"xyz","dob":"12/09/1995","age":"23"}

Эта структура сообщения находится в теме kafka, но я хочу проиндексировать это в эластичном поиске, как показано ниже

{"id":"1","name"{"firstName":"abc","lastName":"xyz"},"dob":"12/09/1995","age":"23"}

как мне этого добиться?

Answer 1

Два варианта:

1. Потоковая обработка данных в теме Kafka.Используя Kafka Streams, вы можете изменять модель данных по мере необходимости.KSQL будет работать для этого в обратном порядке, но пока не поддерживает создание STRUCT.Другие параметры обработки потока: Flink, Spark Streaming и т. Д.

2. Изменение данных при их передаче через Kafka Connect с использованием преобразования одного сообщения.Нет готового преобразования, которое бы это делало, но вы могли бы написать его, используя API .

Отказ от ответственности: я работаю в Confluent, компании, стоящей запроект KSQL с открытым исходным кодом, вклад в Kafka Streams, Kafka Connect и т. д.