elasticsearch - Logstash 和卡夫卡

Question

我的服务器机器上运行着单节点 kafka。
我使用以下命令创建主题“bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test”。
我有两个正在运行的 logstash 实例。第一个从一些java应用程序日志文件中读取数据注入(inject)到kafka。
它工作正常，我可以使用“bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning”命令在控制台上查看kafka中的数据。但是从kafka(相同主题“test”)读取并注入(inject)到elasicsearch的另一个logstash实例失败了。第二个 logstash 实例无法读取数据。我将其配置文件更改为从 kafka 读取并在控制台上打印，然后它也不输出任何内容。这是失败的 logstash 的配置文件:

// config file
     input {
        kafka {
        zk_connect => "localhost:2181"
        topic_id => "test" 
        }
        }
        output {
        stdout{}
        }

Logstash 既不打印任何东西，也不抛出任何错误。
我正在使用 Logstash 2.4 和 kafka 0.10。
我使用了 kafka 快速入门指南 ( http://kafka.apache.org/documentation.html#quickstart)

Answer 1

如果您查看 Kafka 输入 plugin configuration ，你可以看到一个重要的参数，它允许连接到Kafka集群:zk_connect。

根据文档，默认设置为 localhost:2181。确保将其设置为您的 Kafka 集群实例，或者理想情况下是多个实例，具体取决于您的设置。

例如，假设您要连接到具有 JSON 主题的三节点 Kafka 集群。配置如下:

kafka {
 topic_id => "your_topic"
 zk_connect => "kc1.host:2181,kc2.host:2181,kc3.host:2181"
}

此外，为主题配置正确的编解码器也很重要。上面的示例将适用于 JSON 事件。如果使用 Avro，则需要设置另一个参数 - codec .关于如何配置它的详细信息有据可查 on the documentation page .它基本上需要指向 Avro 模式文件，它可以作为 avsc 文件或 Schema Registry 端点给出(在我看来更好的解决方案)。

如果您在 Kafka 环境中运行 Schema Registry，则可以将编解码器指向它的 url。一个完整的例子是:

kafka {
 codec => avro_schema_registry { endpoint => "http://kc1.host:8081"}
 topic_id => "your_topic"
 zk_connect => "kc1.host:2181,kc2.host:2181,kc3.host:2181"
}

希望它有效!