logstash有一个kafka的输入插件。首先，您应该熟悉ApacheKafka及其生产者/消费者范式：https://kafka.apache.org/. 然后开始使用logstash：https://www.elastic.co/products/logstash. 所有这些之后，您将能够使用kafka输入插件进行日志存储：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html. 最后一步是构建一个logstash管道，将数据插入到目标中，比如elasticsearch。这个简单的例子可以帮助您实现目标：
日志存储.conf

input {  
    kafka {
        bootstrap_servers => "localhost:9092"
        topics => ["example-topic"]
    }
}

output {  
    elasticsearch {
        hosts => ["localhost:9200"]
        index => "example-index"
    }
}

在这里，我们只需要从kafka队列中获取特定主题的数据。然后我们将数据存储到elasticsearch索引中。希望有帮助！