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

怼烎@ 2022-08-20 13:10 188阅读 0赞

Flume是Cloudrea公司开源的一款优秀的日志收集框架，主要经历了两个大的版本，分别是  
Flume-OG  
Flume-NG  
OG是0.9.x的版本，依赖zookeeper，角色职责不够单一，  
NG是新版本指1.x的版本，官网解释它更轻量级，更小，角色职责更单一，利用点到点进行容错，当然这也是以后的趋势，  
要理解Flume，就首先理解它的架构，下面看下，官网的一张拓扑图：  
   
  
![3209261e-497c-36d7-89e8-04b48fedfbd9.png][]   
  
名词解释：  
Source：泛指所有的日志收集源，可以是web页面，log文件，数据库，端口，卡口信息等  
Channel：提供中转的临时存储区，可以是本地文件，redis，kakfa，数据库，内存等  
Sink：指日志最终落地的存储区，可以是数据库，HDFS，Hbase，Hive，本地文件等  
Agent：指上面三者组合后的一个完整的数据收集代理，有了这个代理，我们把它安装任何机器上进行收集日志，当然前提是这个Agent符合这个机器上的业务。  
  
   
  
下面看下，安装实战例子：  
  
首先下载flume-ng的安装包：  
  
wget [http://archive.apache.org/dist/flume/1.6.0/apache-flume-1.6.0-bin.tar.gz][http_archive.apache.org_dist_flume_1.6.0_apache-flume-1.6.0-bin.tar.gz]  
  
解压  
tar -zxvf apache-flume-1.6.0-bin.tar.gz  
  
将根目录下的conf下的一些模板文件，重命名成正常文件  
  
  
如果没有配置JAVA\_HOME为环境变量的话，  
则需要在flume-env.sh里面配置jdk的地址  
  
  
1，然后新建一个first.properties的文件，加入以下配置：

Java代码 ![收藏代码][icon_star.png]

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代码 ![收藏代码][icon_star.png]

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代码 ![收藏代码][icon_star.png]

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代码 ![收藏代码][icon_star.png]

1.  \*.\* @@localhost:5140

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

Java代码 ![收藏代码][icon_star.png]

1.  \# Provides TCP syslog reception 
2.  \#$ModLoad imtcp 
3.  \#$InputTCPServerRun 5140

改完之后，重启rsyslog

Java代码 ![收藏代码][icon_star.png]

1.  sudo service rsyslog restart

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

Java代码 ![收藏代码][icon_star.png]

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，几乎大部分功能都能轻松实现，所以，没有谁最好用，谁最不好用，只有谁最适合业务，才是最好的！

[3209261e-497c-36d7-89e8-04b48fedfbd9.png]: /images/20220731/a1e6efb8a54845d69237f36787b87215.png
[http_archive.apache.org_dist_flume_1.6.0_apache-flume-1.6.0-bin.tar.gz]: http://archive.apache.org/dist/flume/1.6.0/apache-flume-1.6.0-bin.tar.gz
[icon_star.png]: /images/20220731/c85a3192e3b540988811097f39444090.png