1， logstash 配置文件
[root@host1: ] cat /opt/logstash/kafka-to-tcp.yml
input {
kafka {
bootstrap_servers => "192.168.0.11:9092" #这里可以是kafka集群，如"192.168.149.101:9092,192.168.149.102:9092"
consumer_threads => 3 #等于 topic分区数
group_id => "logstash_123"
#client_id => "logstash1" #注意，多台logstash实例消费同一个topics时，client_id需要指定不同的名字
#auto_offset_reset => "latest"
auto_offset_reset => "earliest"
topics => ["alertTopic1"]
codec => json { charset => "UTF-8" }
}
}
filter {
#删除某些数据：正则取反，根据json字段ruleName字段内容删除数据
if ([ruleName] !~ ".*主机告警.*") {
drop {}
}
#只保留某些数据：正则匹配，删除其他的数据
#if ([ruleName] =~ ".*主机告警.*") {
# drop {}
#}
mutate {
#删除某些json字段, 修改某些字段内容
remove_field => ["eventId","ruleId"]
gsub => [
"Msg" , "[\r|\n]" , ""
]
}
}
output {
#输出到命令行窗口,方便调试
#stdout{}
#输出到文件，方便排查告警漏告等问题
file {
codec => json_lines { charset => "UTF-8" }
path => "/tmp/b.log"
}
#输出UMP平台对接指定的ip、端口，以指定的格式推送到UMP集中告警平台
tcp {
host => "192.168.0.11"
port => "514"
codec => plain {
format =>"%{TIME} 测试环境--ruleName:%{ruleName},Msg:%{Msg}\n"
}
}
}
2，调试并后台启动
- ./bin/logstash -f /xx/xx.yml
[root@host1: ] cat /usr/lib/systemd/system/logstashtcp.service
[Unit]
Description=Logstash
Requires=network.service
After=network.service
[Service]
LimitNOFILE=65536
LimitMEMLOCK=infinity
WorkingDirectory=/opt/logstash/
ExecStart=/bin/sh bin/logstash -f kafka-to-tcp.yml
ExecReload=/bin/kill -HUP $MAINPID
KillMode=mixed
SuccessExitStatus=143
Restart=on-failure
RestartSec=42s
[Install]
WantedBy=multi-user.target
3, 修改logstash 服务日志路径
sed -i.bak 's@${sys:ls.logs}@/xx/yy@' config/log4j2.properties
重启logstash服务
文章来源地址https://www.toymoban.com/news/detail-708597.html
文章来源：https://www.toymoban.com/news/detail-708597.html
到了这里，关于logstash 消费kafka数据，转发到tcp端口的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
