大数据日志收集框架之Flume入门

Posted 2021-04-30 我是攻城师

Java代码  

1. [webmaster@Hadoop-0-187 conf]$ cat first.properties   

2. # example.conf: A single-node Flume configuration  

3.   

4. # Name the components on this agent  

5. #datasource  

6. a1.sources = r1  

7. #store   

8. a1.sinks = k1  

9. #transfer  

10. a1.channels = c1  

11.   

12. # Describe/configure the source  

13. a1.sources.r1.type = netcat  

14. a1.sources.r1.bind = localhost  

15. a1.sources.r1.port = 44444  

16.   

17. # Describe the sink  

18. a1.sinks.k1.type = logger  

19.   

20. # Use a channel which buffers events in memory  

21. a1.channels.c1.type = memory  

22. a1.channels.c1.capacity = 1000  

23. a1.channels.c1.transactionCapacity = 100  

24.   

25. # Bind the source and sink to the channel  

26. a1.sources.r1.channels = c1  

27. a1.sinks.k1.channel = c1  

配置例子2：收集本地日志到hadoop里面 


hdfs的sink目前支持三种文件类型： 
(1)SequenceFile   文本不可见 
(2)DataStream   设置hdfs.writeFormat=Text，文本可见 
(3)CompressedStream 设置hdfs.codeC=gzip, bzip2, lzo, lzop, snappy，其中一种，后面三个需要编译hadoop时支持才能设置。 

默认是SequenceFile，以文本方式写进入的数据，是不可见的，除非改变类型为DataStream ， 
然后设置hdfs.writeFormat=Text即可，这个选项默认是Writable 

Java代码  

1. a.sources=exec-source  

2. a.sinks=hdfs-sink  

3. a.channels=ch1  

4.   

5. #####source conf  

6.   

7. a.sources.exec-source.type=exec  

8. a.sources.exec-source.command=tail -F /ROOT/server/flume/v.log  

9.   

10.   

11. #####sink conf  

12. a.sinks.hdfs-sink.type=hdfs  

13. a.sinks.hdfs-sink.hdfs.path=hdfs://h1:9000/flume/events  

14. a.sinks.hdfs-sink.hdfs.filePrefix=search  

15. a.sinks.hdfs-sink.hdfs.rollInterval=10  

16. a.sinks.hdfs-sink.hdfs.rollSize=0  

17.   

18. #不设置压缩，指定文本方式加入hdfs  

19. #a.sinks.hdfs-sink.hdfs.fileType=DataStream  

20. #a.sinks.hdfs-sink.hdfs.writeFormat=Text  

21.   

22. #设置压缩lzo或者snappy  

23. a.sinks.hdfs-sink.hdfs.fileType=CompressedStream  

24. #a.sinks.hdfs-sink.hdfs.codeC=snappy  

25. #a.sinks.hdfs-sink.hdfs.codeC=lzo  

26. #a.sinks.hdfs-sink.hdfs.codeC=lzop  

27. #a.sinks.hdfs-sink.hdfs.codeC=gzip  

28. a.sinks.hdfs-sink.hdfs.codeC=bzip2  

29.   

30.   

31. ####channels conf  

32. a.channels.ch1.type=memory  

33. a.channels.ch1.capacity=1000  

34.   

35. a.sources.exec-source.channels=ch1  

36. a.sinks.hdfs-sink.channel=ch1  

启动命令： 最后的a与配置文件里面的agent的name必须一致 
bin/flume-ng agent --conf conf --conf-file  conf/to_hdfs.properties  --name a 

下面是一个按年月日时分，收集的日志： 

Java代码  

1. a.sources=exec-source  

2. a.sinks=hdfs-sink  

3. a.channels=ch1  

4.   

5. #####source conf  

6.   

7. a.sources.exec-source.type=exec  

8. a.sources.exec-source.command=tail -F /ROOT/server/flume/v.log  

9.   

10.   

11. #####sink conf  

12. a.sinks.hdfs-sink.type=hdfs  

13. a.sinks.hdfs-sink.hdfs.path=hdfs://h1:9000/flume/events/%Y/%m/%d/%H/%M  

14. a.sinks.hdfs-sink.hdfs.filePrefix=search  

15. a.sinks.hdfs-sink.hdfs.rollInterval=10  

16. a.sinks.hdfs-sink.hdfs.rollSize=0  

17. a.sinks.hdfs-sink.hdfs.useLocalTimeStamp=true  

18.   

19. #不设置压缩  

20. #a.sinks.hdfs-sink.hdfs.fileType=DataStream  

21. #a.sinks.hdfs-sink.hdfs.writeFormat=Text  

22.   

23. #设置压缩lzo或者snappy  

24. a.sinks.hdfs-sink.hdfs.fileType=CompressedStream  

25. #a.sinks.hdfs-sink.hdfs.codeC=snappy  

26. #a.sinks.hdfs-sink.hdfs.codeC=lzo  

27. #a.sinks.hdfs-sink.hdfs.codeC=lzop  

28. #a.sinks.hdfs-sink.hdfs.codeC=gzip  

29. a.sinks.hdfs-sink.hdfs.codeC=bzip2  

30.   

31.   

32. ####channels conf  

33. a.channels.ch1.type=memory  

34. a.channels.ch1.capacity=1000  

35.   

36. a.sources.exec-source.channels=ch1  

37. a.sinks.hdfs-sink.channel=ch1  

最后看下，配置flume监听linux的rsyslog的日志： 

可用nc模拟telent发送数据到5140端口 
echo "<1>hello via syslog" | nc -t localhost 5140 
配置/etc/rsyslog.conf 
在最后一行，加入tcp转发端口： 

Java代码  

1. *.* @@localhost:5140  

如果不生效，可考虑：取消注释，使下面的生效： 

Java代码  

1. # Provides TCP syslog reception  

2. #$ModLoad imtcp  

3. #$InputTCPServerRun 5140  

改完之后，重启rsyslog 

Java代码  

1. sudo service rsyslog restart  

然后配置flume 
flume监听rsyslog，或者syslog 

Java代码  

1. a.sources=exec-source  

2. a.sinks=hdfs-sink  

3. a.channels=ch1  

4.   

5. #####source conf rsyslog   

6.   

7. a.sources.exec-source.type=syslogtcp  

8. a.sources.exec-source.port=5140  

9. a.sources.exec-source.host=0.0.0.0  

10.   

11.   

12. #####sink conf  

13. a.sinks.hdfs-sink.type=hdfs  

14. a.sinks.hdfs-sink.hdfs.path=hdfs://h1:9000/flume/events/%Y/%m/%d/%H/%M  

15. a.sinks.hdfs-sink.hdfs.filePrefix=search  

16. a.sinks.hdfs-sink.hdfs.rollInterval=10  

17. a.sinks.hdfs-sink.hdfs.rollSize=0  

18. a.sinks.hdfs-sink.hdfs.useLocalTimeStamp=true  

19.   

20. #不设置压缩  

21. #a.sinks.hdfs-sink.hdfs.fileType=DataStream  

22. #a.sinks.hdfs-sink.hdfs.writeFormat=Text  

23.   

24. #设置压缩lzo或者snappy  

25. a.sinks.hdfs-sink.hdfs.fileType=CompressedStream  

26. #a.sinks.hdfs-sink.hdfs.codeC=snappy  

27. #a.sinks.hdfs-sink.hdfs.codeC=lzo  

28. #a.sinks.hdfs-sink.hdfs.codeC=lzop  

29. #a.sinks.hdfs-sink.hdfs.codeC=gzip  

30. a.sinks.hdfs-sink.hdfs.codeC=bzip2  

31.   

32.   

33. ####channels conf  

34. a.channels.ch1.type=memory  

35. a.channels.ch1.capacity=1000  

36.   

37. a.sources.exec-source.channels=ch1  

38. a.sinks.hdfs-sink.channel=ch1  

配置完成启动： 
然后看flume的log再次登陆终端，推出终端，sudo命令执行失败，flume都会采集转发到hdfs上存储起来 


总结： 
Flume不愧是大数据平台的一个标准组件，与Hadoop能非常完美的结合，当然，除了与hadoop结合外，还支持Hbase，Hive等， 相比Logstash+ElasticSearch+Kibana的ELK里面的Logstash，flume比较适合结构化的日志收集，存储，而Logstasch则还可以非常方便的解析，清洗数据，虽然flume通过扩展jar也能支持，但logstash使用的是Jruby语法，相比java，则比较简单。这也是ELK都能够快速部署的原因，当然flume也支持向elasticsearch推送索引数据，通过扩展的jar，几乎大部分功能都能轻松实现，所以，没有谁最好用，谁最不好用，只有谁最适合业务，才是最好的！