elasticsearch - ElasticSearch到Kafka事件-使用Logstash进行的每次更改

Question

最终目标:连接Elasticsearch和kafka并在ES索引中接收正在进行的变更事件到kafka。从kafka，我有听众进行进一步处理。

方法:我使用Logstash输入和输出插件。这是配置代码。

input {
        elasticsearch {
                hosts => ["localhost:9200"]
                index => "liferay-20116"
        }
}
output {
      kafka {
        topic_id => "elastic-topic-index"
        codec => json
      }
}

它正在工作，但是有一个奇怪的问题。

当我听kafka时，它会从ES中读取所有文档，而该文档目前大约有176个文档。

读取后，它会停一会儿，说2秒钟，然后再次读取整个176个文档!

我不知道这是什么问题，这是由于Logstash行为还是Kafka行为异常？

任何帮助，将不胜感激!

Answer 1

这是此插件的标准行为-将数据匹配推送到给定查询。如果只想更改文档，唯一的解决方法是建立对自己所做的更改的了解-例如，您需要为条目添加时间戳，然后将此知识合并到发送给ES的查询中。