csv потоковая передача данных с использованием Kafka

Question

Я пытаюсь отправить данные файла csv через производителя kafka, а затем на стороне потребителя прослушиваю событие.

Producer - командная строка. Я отправляю CSV-файл, используя следующую команду -

kafka-console-producer.bat --broker-list localhost:9092 --topic freshTopic < E:\csv\sample.csv

Я также успешно слушаю мероприятие на стороне потребителя.

Теперь я должен сохранить эти данные в какой-либо базе данных, такой как эластичный поиск. Для этого я должен преобразовать записи CSV в DataModel. Я прочитал учебник ниже, но не в состоянии понять, как я могу написать это в Java. Так может кто-нибудь помочь мне здесь, как я могу преобразовать данные CSV-файла в Datamodel? Заранее спасибо.

Csv потоковая передача данных с использованием kafka

Answer 1

То, что вы написали, будет хорошо работать для передачи данных в Kafka. Есть много способов получить данные в Elasticsearch (который не является базой данных) оттуда ...

Вам не нужен Avro, поскольку JSON тоже будет работать, но confluent-schema-registry не обрабатывает преобразование из CSV, а у Кафки нет класса "DataModel".

Если вы хотите Avro, чтобы поместить его в Elasticsearch как отдельные поля, тогда

1. Вы можете использовать источник сполдира Kafka Connect вместо производителя консоли, и это поможет вам продвинуться дальше, а затем вы сможете запустить соединитель приемника Elasticsearch оттуда
2. Используйте что-то для разбора CSV на Avro, как показывает ваша ссылка (необязательно Python, KSQL тоже может работать)

Если у вас все в порядке с JSON, то Logstash также будет работать