强大的功能,丰富的插件,让logstash在数据处理的行列中出类拔萃
通常日志数据除了要入ES提供实时展示和简单统计外,还需要写入大数据集群来提供更为深入的逻辑处理,前边几篇ELK的文章介绍过利用logstash将kafka的数据写入到elasticsearch集群,这篇文章将会介绍如何通过logstash将数据写入HDFS
本文所有演示均基于logstash 6.6.2版本
数据收集
logstash默认不支持数据直接写入HDFS,官方推荐的output插件是webhdfs
,webhdfs使用HDFS提供的API将数据写入HDFS集群
插件安装
插件安装比较简单,直接使用内置命令即可
# cd /home/opt/tools/logstash-6.6.2
# ./bin/logstash-plugin install logstash-output-webhdfs
配置hosts
HDFS集群内通过主机名进行通信所以logstash所在的主机需要配置hadoop集群的hosts信息
# cat /etc/hosts
192.168.107.154 master01
192.168.107.155 slave01
192.168.107.156 slave02
192.168.107.157 slave03
如果不配置host信息,可能会报下边的错
[WARN ][logstash.outputs.webhdfs ] Failed to flush outgoing items
logstash配置
kafka里边的源日志格式可以参考这片文章:ELK日志系统之使用Rsyslog快速方便的收集Nginx日志
logstash的配置如下:
# cat config/indexer_rsyslog_nginx.conf
input {
kafka {
bootstrap_servers => \"10.82.9.202:9092,10.82.9.203:9092,10.82.9.204:9092\"
topics => [\"rsyslog_nginx\"]
codec => \"json\"
}
}
filter {
date {
match => [\"time_local\",\"dd/MMM/yyyy:HH:mm:ss Z\"]
target => \"time_local\"
}
ruby {
code => \"event.set(\'index.date\', event.get(\'time_local\').time.localtime.strftime(\'%Y%m%d\'))\"
}
ruby {
code => \"event.set(\'index.hour\', event.get(\'time_local\').time.localtime.strftime(\'%H\'))\"
}
}
output {
webhdfs {
host => \"master01\"
port => 50070
user => \"hadmin\"
path => \"/logs/nginx/%{index.date}/%{index.hour}.log\"
codec => \"json\"
}
stdout { codec => rubydebug }
}
logstash配置文件分为三部分:input、filter、output
input指定源在哪里,我们是从kafka取数据,这里就写kafka集群的配置信息,配置解释:
- bootstrap_servers:指定kafka集群的地址
- topics:需要读取的topic名字
- codec:指定下数据的格式,我们写入的时候直接是json格式的,这里也配置json方便后续处理
filter可以对input输入的内容进行过滤或处理,例如格式化,添加字段,删除字段等等
- 这里我们主要是为了解决生成HDFS文件时因时区不对差8小时导致的文件名不对的问题,后边有详细解释
output指定处理过的日志输出到哪里,可以是ES或者是HDFS等等,可以同时配置多个,webhdfs主要配置解释:
- host:为hadoop集群namenode节点名称
- user:为启动hdfs的用户名,不然没有权限写入数据
- path:指定存储到HDFS上的文件路径,这里我们每日创建目录,并按小时存放文件
- stdout:打开主要是方便调试,启动logstash时会在控制台打印详细的日志信息并格式化方便查找问题,正式环境建议关闭
webhdfs还有一些其他的参数例如compression
,flush_size
,standby_host
,standby_port
等可查看官方文档了解详细用法
启动logstash
# bin/logstash -f config/indexer_rsyslog_nginx.conf
因为logstash配置中开了stdout
输出,所以能在控制台看到格式化的数据,如下:
{
\"server_addr\" => \"172.18.90.17\",
\"http_user_agent\" => \"Mozilla/5.0 (iPhone; CPU iPhone OS 10_2 like Mac OS X) AppleWebKit/602.3.12 (KHTML, like Gecko) Mobile/14C92 Safari/601.1 wechatdevtools/1.02.1902010 MicroMessenger/6.7.3 Language/zh_CN webview/ token/e7b92168159736c30401a55589317d8c\",
\"remote_addr\" => \"172.18.101.0\",
\"status\" => 200,
\"http_referer\" => \"https://ops-coffee.cn/wx02935bb29080a7b4/devtools/page-frame.html\",
\"upstream_response_time\" => \"0.056\",
\"host\" => \"ops-coffee.cn\",
\"request_uri\" => \"/api/community/v2/news/list\",
\"request_time\" => 0.059,
\"upstream_status\" => \"200\",
\"@version\" => \"1\",
\"http_x_forwarded_for\" => \"192.168.106.100\",
\"time_local\" => 2019-03-18T11:03:45.000Z,
\"body_bytes_sent\" => 12431,
\"@timestamp\" => 2019-03-18T11:03:45.984Z,
\"index.date\" => \"20190318\",
\"index.hour\" => \"19\",
\"request_method\" => \"POST\",
\"upstream_addr\" => \"127.0.0.1:8181\"
}
查看hdfs发现数据已经按照定义好的路径正常写入
$ hadoop fs -ls /logs/nginx/20190318/19.log
-rw-r--r-- 3 hadmin supergroup 7776 2019-03-18 19:07 /logs/nginx/20190318/19.log
至此kafka到hdfs数据转储完成
遇到的坑
HDFS按小时生成文件名不对
logstash在处理数据时会自动生成一个字段@timestamp
,默认情况下这个字段存储的是logstash收到消息的时间,使用的是UTC时区,会跟国内的时间差8小时
我们output到ES或者HDFS时通常会使用类似于rsyslog-nginx-%{+YYYY.MM.dd}
这样的变量来动态的设置index或者文件名,方便后续的检索,这里的变量YYYY
使用的就是@timestamp
中的时间,因为时区的问题生成的index或者文件名就差8小时不是很准确,这个问题在ELK架构中因为全部都是用的UTC时间且最终kibana展示时会自动转换我们无需关心,但这里要生成文件就需要认真对待下了
这里采用的方案是解析日志中的时间字段time_local
,然后根据日志中的时间字段添加两个新字段index.date
和index.hour
来分别标识日期和小时,在output的时候使用这两个新加的字段做变量来生成文件
logstash filter配置如下:
filter {
# 匹配原始日志中的time_local字段并设置为时间字段
# time_local字段为本地时间字段,没有8小时的时间差
date {
match => [\"time_local\",\"dd/MMM/yyyy:HH:mm:ss Z\"]
target => \"time_local\"
}
# 添加一个index.date字段,值设置为time_local的日期
ruby {
code => \"event.set(\'index.date\', event.get(\'time_local\').time.localtime.strftime(\'%Y%m%d\'))\"
}
# 添加一个index.hour字段,值设置为time_local的小时
ruby {
code => \"event.set(\'index.hour\', event.get(\'time_local\').time.localtime.strftime(\'%H\'))\"
}
}
output的path中配置如下
path => \"/logs/nginx/%{index.date}/%{index.hour}.log\"
HDFS记录多了时间和host字段
在没有指定codec的情况下,logstash会给每一条日志添加时间和host字段,例如:
源日志格式为
ops-coffee.cn | 192.168.105.91 | 19/Mar/2019:14:28:07 +0800 | GET / HTTP/1.1 | 304 | 0 | - | 0.000
经过logstash处理后多了时间和host字段
2019-03-19T06:28:07.510Z %{host} ops-coffee.cn | 192.168.105.91 | 19/Mar/2019:14:28:07 +0800 | GET / HTTP/1.1 | 304 | 0 | - | 0.000
如果不需要我们可以指定最终的format只取message,解决方法为在output中添加如下配置:
codec => line {
format => \"%{message}\"
}
同时output到ES和HDFS
在实际应用中我们需要同时将日志数据写入ES和HDFS,那么可以直接用下边的配置来处理
# cat config/indexer_rsyslog_nginx.conf
input {
kafka {
bootstrap_servers => \"localhost:9092\"
topics => [\"rsyslog_nginx\"]
codec => \"json\"
}
}
filter {
date {
match => [\"time_local\",\"dd/MMM/yyyy:HH:mm:ss Z\"]
target => \"@timestamp\"
}
ruby {
code => \"event.set(\'index.date\', event.get(\'@timestamp\').time.localtime.strftime(\'%Y%m%d\'))\"
}
ruby {
code => \"event.set(\'index.hour\', event.get(\'@timestamp\').time.localtime.strftime(\'%H\'))\"
}
}
output {
elasticsearch {
hosts => [\"192.168.106.203:9200\"]
index => \"rsyslog-nginx-%{+YYYY.MM.dd}\"
}
webhdfs {
host => \"master01\"
port => 50070
user => \"hadmin\"
path => \"/logs/nginx/%{index.date}/%{index.hour}.log\"
codec => \"json\"
}
}
这里我使用logstash的date插件将日志中的\"time_local\"字段直接替换为了@timestamp,这样做有什么好处呢?
logstash默认生成的@timestamp字段记录的时间是logstash接收到消息的时间,这个时间可能与日志产生的时间不同,而我们往往需要关注的时间是日志产生的时间,且在ELK架构中Kibana日志输出的默认顺序就是按照@timestamp来排序的,所以往往我们需要将默认的@timestamp替换成日志产生的时间,替换方法就用到了date插件,date插件的用法如下
date {
match => [\"time_local\",\"dd/MMM/yyyy:HH:mm:ss Z\"]
target => \"@timestamp\"
}
match:匹配日志中的时间字段,这里为time_local
target:将match匹配到的时间戳存储到给定的字段中,默认不指定的话就存到@timestamp字段
另外还有参数可以配置:timezone
,locale
,tag_on_failure
等,具体可查看官方文档
如果你觉得文章不错,请点右下角【在看】。如果你觉得读的不尽兴,推荐阅读以下文章: