Logstash插件介绍

Logstash插件介绍

在生产环境中，Logstash 的管道要复杂很多，可能需要配置多个输入、过滤器和输出插件。因此，需要一个配置文件管理输入、过滤器和输出相关的配置。配置文件内容格式如下：

#　输入
input {
...
}
# 过滤器
filter {
...
}
# 输出
output {
...
}

根据自己的需求在对应的位置配置输入插件、过滤器插件、输出插件和编码解码插件即可。

在使用插件之前，我们先了解一个概念：事件。Logstash 每读取一次数据的行为叫做事件。在 Logstach 目录中创建一个配置文件，名为 logstash.conf（名字任意）。

1、输入插件

输入插件允许一个特定的事件源可以读取到 Logstash 管道中，配置在 input {} 中，且可以设置多个。

新建配置文件：

vi demo1.conf

input {
# 从文件读取日志信息
file {
# 文件路径
path => "/var/log/anaconda.log"
start_position => "beginning" }
}
# filter {
#
# }
output {
# 标准输出
stdout { codec => rubydebug }
}

执行：

bin/logstash -f config/demo1.conf

2、数据写入到ES中

新建配置文件：

vi  data2es.conf

input {
file {
type => "game"
path => "/root/log.json/"
discover_interval => 10
start_position => "beginning"
}
}
output {
elasticsearch {
index => "game-%{+YYYY.MM.dd}"
hosts => ["192.168.52.140:9200"]
}
}

执行：

bin/logstash -f config/data2es.conf

3、读取kafka中topic数据

(1) 启动zookeeper kafka集群

(2) 创建主题:

kafka-topics.sh --zookeeper node-1:2181 --create --topic accesslog --replication-factor 1 --partitions 3

(3)执行生产者脚本:

kafka-console-producer.sh --broker-list node-1:9092 --topic accesslog

执行消费者脚本，查看是否写入:

kafka-console-consumer.sh --zookeeper node-1:2181 --from-beginning --topic accesslog

(4)新建logstash配置文件

vi readkafka.conf

input{
kafka {
#介绍:plain是一个空的解析器，它可以让用户自己指定格式
codec => "plain"
group_id => "logstash1"
auto_offset_reset => "earliest"
topics  => ["hello"]
bootstrap_servers => ["node-1:9200"] # broker的地址
}

}
output{
stdout{
codec => rubydebug
}
}

执行：

bin/logstash -f config/readkafka.conf

4、数据写入到kafka中

新建配置文件：

vi  data2kafka.conf

input {
file {
path => "/root/access.log"
discover_interval => 5
start_position => "beginning"
}
}
output {
kafka {
topic_id => "accesslog"
bootstrap_servers => ["node-1:9092,node-2:9092,node-3:9092"]
}
}

执行：

bin/logstash -f config/data2kafka.conf

喜欢就点赞评论+关注吧



感谢阅读，希望能帮助到大家，谢谢大家的支持！