Logstash-Kafka 集成不起作用

Question

我们想使用 logstash 获取日志并将它们传递给 kafka。

我们为 logstash 1.5.0 beta 1 和 kafka 2.9.2_0.8.1.1 编写了以下 conf 文件

**

input {    
         file {    
         type => "apache"    
         path => ["/var/log/apache2/access.log", "/var/log/apache2/error.log"]    
     }
}




output {    
    kafka {    
        codec => plain {    
            format => "%{message}"    
        }
    topic_id => "example1"    
    }    
}

**

运行以下命令后：bin/logstash agent -f test.conf --log ex.log

test.conf 是我们的配置文件。 ex.log 是我们为存储日志创建的空白文件。

我们得到以下输出

将 logstash 日志发送到 ex.log。 使用里程碑 2 输入插件“文件”。这个插件应该是稳定的，但如果你看到奇怪的行为，请告诉我们！有关插件里程碑的更多信息，请参阅http://logstash.net/docs/1.5.0.beta1/plugin-milestones {:level=>:warn} 使用里程碑 1 输出插件“kafka”。这个插件应该可以工作，但会受益于像你这样的人的使用。如果您发现错误或对如何改进此插件有建议，请告诉我们。有关插件里程碑的更多信息，请参阅http://logstash.net/docs/1.5.0.beta1/plugin-milestones {:level=>:warn} log4j:WARN 找不到记录器 (kafka.utils.VerifiableProperties) 的附加程序。 log4j:WARN 请正确初始化 log4j 系统。 log4j:WARN 有关详细信息，请参阅 http://logging.apache.org/log4j/1.2/faq.html#noconfig。 SLF4J：无法加载类“org.slf4j.impl.StaticLoggerBinder”。 SLF4J：默认为无操作（NOP）记录器实现 SLF4J：有关详细信息，请参阅 http://www.slf4j.org/codes.html#StaticLoggerBinder。

我们尝试在 bashrc 中设置 CLASSPATH。不工作。 请告诉我们哪里出错了。 提前谢谢你！

Answer 1

您可以检查的一些事项： 主题 example1 可用吗？如果没有，您是否在kafka中使用了自动创建？像这样检查现有的主题：

bin/kafka-topics.sh --list --zookeeper localhost:2181

Example1 应该在返回的项目之间，如果不是你也可以手动创建主题。

bin/kafktopics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic example1

查看kafka是否收到消息启动消费者：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic example1

我启动了一个简单的 logstash 实例，它监听标准并使用消费者查看是否有消息到达 kafka。这是我的 logstash 配置：

input {
        stdin { }
}

output {
    kafka {
        codec => plain {
            format => "%{message}"
        }
        topic_id => "example1"
    }
}

希望有帮助