logstash將Kafka中的日誌數據訂閱到HDFS

前言：一般狀況下，咱們將Kafka的日誌數據經過logstash訂閱輸出到ES，而後用Kibana來作可視化分析，這就是咱們一般用的ELK日誌分析模式。可是基於ELK的日誌分析，一般比較經常使用的是實時分析，日誌存個十天半個月都會刪掉。那麼在一些狀況下，我須要將日誌數據也存一份到我HDFS，積累到比較久的時間作半年、一年甚至更長時間的大數據分析。下面就來講如何最簡單的經過logstash將kafka中的數據訂閱一份到hdfs。

一：安裝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

已經能夠成功寫入了。