配置logstash消费kafka多个topic，分别⽣成索引filebeat配置多个topic
#filebeat.prospectors:
filebeat.inputs:
- input_type: log
encoding: GB2312
#  fields_under_root: true
fields:  ##添加字段
serverip: 192.168.1.10
logtopic: wap
enabled: True
paths:
-
安徽茶叶 /app/wap/logs/catalina.out
multiline.pattern: '^\['  #java报错过滤
multiline.match: after
tail_files: false
- input_type: log
encoding: GB2312
# fields_under_root: true
fields:  ##添加字段
serverip: 192.168.1.10
英文游戏名称logtopic: api
enabled: True
paths:
- /app/api/logs/catalina.out
multiline.pattern: '^\['  #java报错过滤
multiline.match: after
tail_files: false
#----------------------------- Logstash output --------------------------------
output.kafka:
enabled: true
hosts: ["192.168.16.222:9092","192.168.16.237:9092","192.168.16.238:9092"]
河蚌怎么做好吃topic: 'elk-%{[fields.logtopic]}' ##匹配fileds字段下的logtopic
partition.hash:
reachable_only: true
compression: gzip
max_message_bytes: 1000000
required_acks: 1
<_files: true北京旅游景点大全
查看是否输出到kafka
$  bin/kafka-topics.sh --list --zookeeper kafka-01:2181, kafka-02:2181,kafka-03:2181
elk-wap
elk-api
配置logstash集
input{
kafka{
bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"
topics_pattern  => "elk-.*"
consumer_threads => 5
decorate_events => true
codec => "json"
auto_offset_reset => "latest"
group_id => "logstash1"##logstash 集需相同
}
}
王鹤棣女朋友刘艺分手了吗filter {
ruby {
code => "event.timestamp.time.localtime"
}
mutate {
remove_field => ["beat"]
}
grok {
match => {"message" => "\[(?<time>\d+-\d+-\d+\s\d+:\d+:\d+)\] \[(?<level>\w+)\] (?<thread>[\w|-]+) (?<class>[\w|\.]+) (?<lineNum>\d+):(?<msg>.+)" }
}
}
output {
elasticsearch {
hosts => ["192.168.16.221:9200","192.168.16.251:9200","192.168.16.252:9200"]
#      index => "%{[fields][logtopic}" ##直接在⽇志中匹配，索引会去掉elk
index =>  "%{[@metadata][topic]}-%{+YYYY-MM-dd}"
}
stdout {
codec => rubydebug
新四小花旦}
logstash集配置
⼀机多实例，同⼀个配置⽂件，启动时只需更改数据路径
./bin/logstash -f --path.data=/usr/local/logdata/
多台机器
logstash配置⽂件group_id 相同即可