Logstash安装部署配置

为了实现各业务平台日志信息采集到大数据平台hdfs上。之前的定技术栈是 flume->kafka->storm->hdfs. 其中通过storm需要写代码，稳定性，可扩展性，维护性不好。

从kafka到hdfs，有专门的日志工具logstash可以解决这个问题. 目前已经在我们的开发环境稳定的运行了一个礼拜（http://192.168.23.31:50070/explorer.html#/data/logstash）。

现请部署到生产环境。

logstash安装配置如下：

1. logstash下载安装, 下载比较慢的话，可以传给你。

wget -c https://download.elastic.co/logstash/logstash/packages/centos/logstash-2.3.4-1.noarch.rpm

rpm -ivh logstash-2.3.4-1.noarch.rpm

2. logstash hdfs插件下载安装

git clone https://github.com/heqin5136/logstash-output-webhdfs-discontinued.git

cd logstash-output-webhdfs-discontinued

/opt/logstash/bin/plugin install logstash-output-webhdfs

3.logstash配置

vim /etc/logstash/conf.d/logstash.conf

input {

kafka {

zk_connect =>"192.168.1.50:2181,192.168.1.51:2181,192.168.1.52:2181" #kafka的zk集群地址，请改为生产环境

group_id => "hdfs" #消费者组，不要和ELK上的消费者一样

topic_id => "flume_kafka_channel_topic" #kafka topic，改为生产环境

consumer_id => "logstash-consumer-192.168.23.31" #消费者id，自定义

consumer_threads => 1

queue_size => 200

codec => plain{ charset => "UTF-8" }

auto_offset_reset => "smallest"

}

}

filter {

grok {

match => { "message" =>

#"%{TIMESTAMP_ISO8601:date} (?<thread_name>.+?\bhost\b.+?)(?<thread>.+?\bu001Cbi\b)(?<action>.+?\baction\b)(?<type>.+?\btype\b)(?<content>.*)"

"(?<thread>.+?\bu001Cbi\b)(?<action>.+?\baction\b)(?<type>.+?\btype\b)(?<content>.*)"

}

}

}

output {

#如果你一个topic中会有好几种日志，可以提取出来分开存储在hdfs上。

if [action] == "-action" and [type] == "-type" {

webhdfs {

workers => 2

host => "192.168.23.31" #hdfs的namenode地址，改为生产环境

port => 50070 #webhdfs端口

user => "root" #hdfs运行的用户啊，以这个用户的权限去写hdfs。

path => "/data/logstash/log-%{+YYYY}-%{+MM}/apiLog-%{+YYYY}-%{+MM}-%{+dd}.log" #按月建目录，按天建log文件。

flush_size => 500

#compression => "snappy" #压缩格式，可以不压缩

idle_flush_time => 10

retry_interval => 0.5

codec => plain{ charset => "UTF-8" }

}

}

}

4. logstash 配置检查，启动，停止

/etc/init.d/logstash configtest start stop