准备

• JDK 1.8
• Hadoop 3.3.0
• logstash 7.2.0
• filebeat 7.2.0

以上安装简单，此处不再做介绍，hadoop集群安装参考

filebeat配置

此处仅为简要配置，详细信息请参考官方文档

filebeat.inputs:
- type: log  # 收集数据类型 log、json
  # 是否开启收集
  enabled: true
  # 收集日志的路径，可模糊匹配
  paths:
    - /data/logs/test/*.log
  # 可以设置标签，接收数据时根据标签区分处理
  tags: ["test"] 
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 1
setup.kibana:
output.logstash:
  # logstash 地址，可以多个
  hosts: ["logstash.dongle.com:4560"]
processors:
- add_host_metadata: ~

logstash配置

此处仅为简要配置，详细信息请参考官方文档
编辑配置文件 $LOGSTASH_HOME/config/test.conf（自定义）

input { 
	# 接收输入端口
    beats{ port => 4560}
}
filter{
# 对一些参数做处理
        ruby{
                code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
        }
        ruby{
                code => "event.set('@timestamp',event.get('timestamp'))"
        }
        mutate{
                remove_field => ["timestamp"]
        }
}

启动Filebeat

${FILEBEAT_HOME}/bin/filebeat

logstash同步hadoop方式一：webhdfs方式

output {
# 数据输出
      webhdfs{
              host=>"hadoop.dongle.com"
              port => 9870
              user => "root"
              path => "/data/logs/test/logstash-%{+YYYY-MM-dd-HH}.log"
              codec => "line { format => '%{[@metadata][ip_address]}|%{message}'}"
              retry_interval => 3
              retry_times => 3
      }
}

logstash同步hadoop方式二：本地落盘方式和Hadoop上传

因webhdfs是通过web接口传递数据，容易不稳定，可以先落盘再上传，方便后期失败修复

output{
      file{
              path => "/data/logs/test/logstash-%{+YYYY-MM-dd-HH}.log"
              codec => line { format => "%{[host][ip][0]}|%{message}"}
      }
}

hadoop 上传本地文件到hdfs

hadoop fs -put '/data/logs/test/logstash-2021-01-01-11.log' /data/logs/test/

启动logstash

${LOGSTASH_HOME}/bin/logstash -f config/test.conf

hadoop查看hdfs

hadoop fs -ls /data/logs/test/