Logstash讀取Kafka數據寫入HDFS詳解

強大的功能，豐富的插件，讓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，那么可以直接用下邊的配置來處理

文章未完，全部內容請關注公眾號【運維咖啡吧】或個人網站https://ops-coffee.cn查看，運維咖啡吧專注於原創精品內容分享，感謝您的支持