flume NG集中处理日志

Dear 丶 2022-07-13 14:51 131阅读 0赞

### 1、flume NG安装 ###

具体下载地址http://archive.apache.org/dist/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz

安装解压flume安装包，命令如下所示：

1.  tar -zxvf apache-flume-1.7.0\-bin.tar.gz

配置环境变量

1.  export FLUME\_HOME=/home/hadoop/cloud/programs/flume
2.  export PATH=$PATH:$FLUME\_HOME/bin

### 2、flume NG的配置 ###

开始之前，先约定产生日志我们称为client端，存储日志的我们称为server端。

无论是client端还是server端，都分为source channel sink三个部分。

只是client端的sink连接的是server的source。大致是这个样子。

![A fan-in flow using Avro RPC to consolidate events in one place][]

配置文件的主要目的，就是配置一条链路，一个链路称为一个agent，一个agent需要配置source，channel，sink三个组件，每个组件的具体配置flume ng官网有，下面配置示例是文件转到hdfs，中转的chennel也是文件

1.  \#agent1 name 
2.  agent1.sources=source1 
3.  agent1.sinks=sink1 
4.  agent1.channels=channel1 
5.  \#Spooling Directory 
6.  \#set source1 
7.  agent1.sources.source1.type=spooldir 
8.  agent1.sources.source1.spoolDir=/home/hadoop/flumetest/dir/logdfs 
9.  agent1.sources.source1.channels=channel1 
10. agent1.sources.source1.fileHeader = false
11. agent1.sources.source1.interceptors = i1 
12. agent1.sources.source1.interceptors.i1.type = timestamp 
13. \#set sink1 
14. agent1.sinks.sink1.type=hdfs 
15. agent1.sinks.sink1.hdfs.path=hdfs://hadoopmaster:8020/flume/logdfs
16. agent1.sinks.sink1.hdfs.fileType=DataStream 
17. agent1.sinks.sink1.hdfs.writeFormat=TEXT 
18. agent1.sinks.sink1.hdfs.rollInterval=1
19. agent1.sinks.sink1.channel=channel1 
20. agent1.sinks.sink1.hdfs.filePrefix=%Y-%m-%d 
21. agent1.sinks.sink1.hdfs.fileSuffix=.txt 
22. \#set channel1 
23. agent1.channels.channel1.type=file 
24. agent1.channels.channel1.checkpointDir=/home/hadoop/flumetest/dir/logdfstmp/point 
25. agent1.channels.channel1.dataDirs=/home/hadoop/flumetest/dir/logdfstmp

### 启动 ###

1.  flume-ng agent --conf conf --conf-file /home/hadoop/cloud/programs/flume/conf/flume-hdfs.conf --name agent1 -Dflume.root.logger=INFO,console > /home/hadoop/cloud/programs/flume/logs/flume-hdfs.log 2>&1 &

### 集群配置    ###

理解单点配置的原理以后，配置集群就不难理解了。

接着上例：

file source ->file chennel->hdfs1/hdfs2

source和chennel的配置同上，sink的配置要配置一个sink group

具体见下：

1.  \#agent1 name 
2.  agent1.channels = c1 
3.  agent1.sources = r1 
4.  agent1.sinks = k1 k2 
5.  \#set gruop 
6.  agent1.sinkgroups = g1 
7.  \#set channel 
8.  agent1.channels.c1.type = memory 
9.  agent1.channels.c1.capacity = 1000
10. agent1.channels.c1.transactionCapacity = 100
11. agent1.sources.r1.channels = c1 
12. agent1.sources.r1.type = exec 
13. agent1.sources.r1.command = tail -F /home/hadoop/flumetest/dir/logdfs/flumetest.log 
14. agent1.sources.r1.interceptors = i1 i2 
15. agent1.sources.r1.interceptors.i1.type = static
16. agent1.sources.r1.interceptors.i1.key = Type 
17. agent1.sources.r1.interceptors.i1.value = LOGIN 
18. agent1.sources.r1.interceptors.i2.type = timestamp 
19. \# set sink1 
20. agent1.sinks.k1.channel = c1 
21. agent1.sinks.k1.type = avro 
22. agent1.sinks.k1.hostname = hadoopmaster 
23. agent1.sinks.k1.port = 52020
24. \# set sink2 
25. agent1.sinks.k2.channel = c1 
26. agent1.sinks.k2.type = avro 
27. agent1.sinks.k2.hostname = hadoopslave1 
28. agent1.sinks.k2.port = 52020
29. \#set sink group 
30. agent1.sinkgroups.g1.sinks = k1 k2 
31. \#set failover 
32. agent1.sinkgroups.g1.processor.type = failover 
33. agent1.sinkgroups.g1.processor.priority.k1 = 10
34. agent1.sinkgroups.g1.processor.priority.k2 = 1
35. agent1.sinkgroups.g1.processor.maxpenalty = 10000

[A fan-in flow using Avro RPC to consolidate events in one place]: /images/20220713/b3ff73abb1b5479d841b59f8507e4da3.png