Logstash将Kafka中的日志数据订阅到HDFS

374次阅读

共计 2520 个字符，预计需要花费 7 分钟才能阅读完成。

前言：通常情况下，我们将 Kafka 的日志数据通过 logstash 订阅输出到 ES，然后用 Kibana 来做可视化分析，这就是我们通常用的 ELK 日志分析模式。但是基于 ELK 的日志分析，通常比较常用的是实时分析，日志存个十天半个月都会删掉。那么在一些情况下，我需要将日志数据也存一份到我 HDFS，积累到比较久的时间做半年、一年甚至更长时间的大数据分析。下面就来说如何最简单的通过 logstash 将 kafka 中的数据订阅一份到 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 将 Kafka 中的日志数据订阅到 HDFS

四：配置 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}
}