使用kafka向elasticsearch导入数据，如何获取导入数据的进度以及错误的日志？

Question

我们目前采用的数据导入的框架是这样的：

但是consumer是我们用kafka-python写的python代码，当程序崩溃的时候，有保护机制重新启动，但是有时会重复导入相同数据。
同时这个consumer还要更新数据导入的进度到es以及收集错误的日志。

想向大家咨询一下，大家一般都是用什么方式来确保数据完整，以及如何获取数据导入进度和错误日志的呢？

比如有没有使用kafka-connect-elasticsearch 或者 spark streaming？

Answer 1

这里同时，同步一个我在stackoverflow上的提问， 虽然给答案加了一分，但是目前回答并没有解决进度和错误日志的问题。地址： https://stackoverflow.com/que...