logstash将Kafka中的日志数据订阅到HDFS
前言:通常情况下,我们将Kafka的日志数据通过logstash订阅输出到ES,然后用Kibana来做可视化分析,这就是我们通常用的ELK日志分析模式。但是基于ELK的日志分析,通常比较常用的是实时分析,日志存个十天半个月都会删掉。那么在一些情况下,我需要将日志数据也存一份到我HDFS,积累到比较久的时间做半年、一年甚至更长时间的大数据分析。下面就来说如何最简单的通过logstash将kafka中的数据订阅一份到hdfs。
创新互联主要从事成都网站建设、做网站、网页设计、企业做网站、公司建网站等业务。立足成都服务昭通,十余年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:18982081108
一:安装logstash(下载tar包安装也行,我直接yum装了)
#yum install logstash-2.1.1
二:从github上克隆代码
#git clone https://github.com/heqin5136/logstash-output-webhdfs-discontinued.git #ls logstash-output-webhdfs-discontinued
三:安装logstash-output-webhdfs插件
#cd logstash-output-webhdfs-discontinued logstash的bin目录下有个plugin,使用plugin来安装插件 #/opt/logstash/bin/plugin install logstash-output-webhdfs
四:配置logstash
#vim /etc/logstash/conf.d/logstash.conf input { kafka { zk_connect => '10.10.10.1:2181,10.10.10.2:2181,10.10.10.3:2181' #kafka的zk集群地址 group_id => 'hdfs' #消费者组,不要和ELK上的消费者一样 topic_id => 'apiAppWebCms-topic' #topic consumer_id => 'logstash-consumer-10.10.8.8' #消费者id,自定义,我写本机ip。 consumer_threads => 1 queue_size => 200 codec => 'json' } } output { #如果你一个topic中会有好几种日志,可以提取出来分开存储在hdfs上。 if [type] == "apiNginxLog" { webhdfs { workers => 2 host => "10.10.8.1" #hdfs的namenode地址 port => 50070 #webhdfs端口 user => "hdfs" #hdfs运行的用户啊,以这个用户的权限去写hdfs。 path => "/data/logstash/apiNginxLog-%{+YYYY}-%{+MM}-%{+dd}/logstash-%{+HH}.log #按天建目录,按小时建log文件。 flush_size => 500 # compression => "snappy" #压缩格式,可以不压缩 idle_flush_time => 10 retry_interval => 0.5 } } if [type] == "apiAppLog" { webhdfs { workers => 2 host => "10.64.8.1" port => 50070 user => "hdfs" path => "/data/logstash/api/apiAppLog-%{+YYYY}-%{+MM}-%{+dd}.log" flush_size => 500 # compression => "snappy" idle_flush_time => 10 retry_interval => 0.5 } } stdout { codec => rubydebug } }
五:启动logstash
#/etc/init.d/logstash start
已经可以成功写入了。
分享题目:logstash将Kafka中的日志数据订阅到HDFS
网页链接:http://scyanting.com/article/goiodd.html