logstash 消费kafka数据，转发到tcp端口

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
ExecStart=/opt/logstash/bin/logstash -f /opt/logstash/kafka-to-tcp.yml
ExecReload=/bin/kill -HUP $MAINPID
KillMode=mixed
SuccessExitStatus=143
Restart=on-failure
RestartSec=42s[Install]
WantedBy=multi-user.target