将数据从kafka导入elasticsearch时如何获取导入进度和错误日志？

Question

背景：

我正在尝试将数据从 kafka 导入到 elasticsearch，并且有 2 种客户端。一种是网络客户端，另一种是代理客户端。

Web 客户端会在用户上传时处理 csv 文件，Web 客户端每 10,000 行从 csv 文件中读取一次，并将带有 csv 总行数的数据消息发送给 Producer。 Producer将消息发送给kafka，然后consumer拉取消息，并将数据导入elasticsearch。同时消费者使用数据消息长度和 csv 总计数来更新任务进度，如果有，也会更新错误日志。最后我们的网络客户端会知道错误和导入进度。

Agent 客户端监视日志文件的变化，一旦有新的日志到来，它会向生产者发送消息，与 Web 客户端相同，但它不关心进度。因为日志总是像 nginx 日志一样增长。

框架：

这是我使用的框架：

生产者和消费者是我们使用kafka-python的python程序。

问题：

1. 有时消费者会崩溃，它会自动重启和 再次重新导入相同的数据。
2. 有时客户端发送太多 消息，生产者可能会错过一些，因为 http 请求有 我猜是限制。

问题：

有没有更好的框架来做这些事情？喜欢使用 kafka-connect-elasticsearch ，火花流？

Answer 1

是的 - 使用 Kafka Connect Elasticsearch connector。这会让你的生活轻松很多。 Kafka Connect API 专门为您完成所有这些困难的事情（重启、偏移管理等）。作为最终用户，您只需要设置一个配置文件。您可以阅读使用 Kafka Connect here 的示例。

Kafka Connect 是 Apache Kafka 的一部分。 Elasticsearch 连接器是开源的，可在github 上单独使用。或者，只需下载Confluent Platform，它将最新版本的 Kafka 与连接器（包括 Elasticsearch、HDFS 等）和一堆其他有用的工具捆绑在一起。