Logstash subscribes log data from Kafka to HDFS

In general, we output the log data of Kafka to ES through logstash subscription, and then use Kibana to do visual analysis, which is the ELK log analysis mode we usually use. However, the log analysis based on ELK is usually used in real-time analysis, and the log will be deleted after ten days and a half months. Well, in some cases, I need to save a copy of the log data to my HDFS and accumulate it for a long time to do big data analysis for half a year, a year or even longer. Here's the easiest way to subscribe data from kafka to hdfs through logstash.

One: install logstash (download tar package and install it, I installed it directly with yum)

# yum install logstash-2.1.1

Two: clone the code from github

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

Three: install the logstash-output-webhdfs plug-in

There is a plugin in the bin directory of # cd logstash-output-webhdfs-discontinuedlogstash. Use plugin to install the plug-in # / opt/logstash/bin/plugin install logstash-output-webhdfs

Four: configure logstash

# vim / etc/logstash/conf.d/logstash.confinput {kafka {zk_connect = > '10.10.10.1 kafka 2181, 10.10.10.2 zk_connect 2181, 10.10.10.3 kafka 2181 # kafka zk cluster address group_id = >' hdfs' # Consumer Group Don't be like the consumer on ELK topic_id = > 'apiAppWebCms-topic' # topic consumer_id = >' logstash-consumer-10.10.8.8' # Consumer id, custom, I write native ip. Consumer_threads = > 1 queue_size = > 200 codec = > 'json'}} output {# if you have several logs in a topic, you can extract them and store them separately on the hdfs. If [type] = = "apiNginxLog" {webhdfs {workers = > 2 host = > namenode address of "10.10.8.1" # hdfs port = > 50070 # webhdfs port user = > "hdfs" # hdfs running user, with this user's permission to write hdfs. Path = > "/ data/logstash/apiNginxLog-% {+ YYYY} -% {+ MM} -% {+ dd} / logstash-% {+ HH} .log # build directories by day and log files by hour. Flush_size = > 50 compressed compression = > "snappy" # compressed format 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 = > 50 years compression = > "snappy" idle_flush_time = > 10 retry_interval = > 0.5}} stdout {codec = > rubydebug}}

Five: start logstash

# / etc/init.d/logstash start

It can be successfully written.