Возможно ли с Kafka подключиться к ElasticSearch

Question

Я получаю события, которые попадают в Кафку. Из этих событий я беру идентификатор с помощью приложения Kafka Streams и отправляю его обратно в Kafka как пару (id, 1) в другой теме. Затем я хотел бы узнать, существует ли идентификатор уже в ElasticSearch, и если так обновить его счетчик, в противном случае создайте новую запись в ElasticSearch с идентификатором от Kafka и счетчиком, установленным в 1, то есть с обновлением записи (id, 1) пальцы.

Я надеялся использовать Kafka Connect для ElasticSearch для этого, но, похоже, это не так просто, если возможно вообще. Я вижу, что добавление записей в ES работает, но слияние с существующими записями кажется тем, о чем я еще не узнал. Возможно ли это уже, и если да, то как, а если нет, планируется ли это в ближайшем выпуске?

Answer 1

Я разветвил datamountaineer ES sink connector, чтобы позволить Upsert. С его помощью вы можете указать ПК и запустить обновление с помощью docAsUpsert в ES. Вы можете захватить проект и скомпилировать Jar от my github fork.