flume使用之exec source收集各端数据汇总到另外一台服务器

小鱼儿 2021-05-12 11:55 292阅读 0赞

一、场景一描述：

线上api接口服务通过log4j往本地磁盘上打印日志，在接口服务器上安装flume，通过exec source收集日志，然后通过avro sink发送到汇总服务器上的flume；汇总服务器上的flume通过avro source接收日志，然后通过file\_roll sink写到本地磁盘。

![Center][]

假设：api接口服务器两台 10.153.140.250和10.153.140.251，汇总日志的服务器一台 10.153.137.211

1、api接口服务器上flume配置：

1）在api接口服务器上下载、解压、安装flume：

cd /usr/local/  
    wget http://mirror.bit.edu.cn/apache/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz  
    tar -xvzf apache-flume-1.7.9-bin.tar.gz  
    
    vim /etc/profile  
      
    export PS1="[\u@`/sbin/ifconfig eth0|grep 'inet '|awk -F'[: ]+' '{print $4}'` \W]"'$ '  
    export FLUME_HOME=/usr/local/apache-flume-1.6.0-bin  
    export PATH=$PATH:$FLUME_HOME/bin

2）修改flume-env.sh 配置文件:

cd /usr/local/flume/conf  
    vim flume-env.sh

里面指定java\_home，同时在conf目录添加log4j.properties文件；

3）flume配置文件：

agent1.sources = ngrinder
    agent1.channels = mc1
    agent1.sinks = avro-sink
    
    agent1.sources.ngrinder.type = exec
    agent1.sources.ngrinder.command = tail -F /data/logs/ttbrain/ttbrain-recommend-api.log
    agent1.sources.ngrinder.channels = mc1
    
    agent1.channels.mc1.type = memory
    agent1.channels.mc1.capacity = 1000
    agent1.channels.mc1.keep-alive = 60
    
    agent1.sinks.avro-sink.type = avro
    agent1.sinks.avro-sink.channel = mc1
    agent1.sinks.avro-sink.hostname = 10.153.137.211
    agent1.sinks.avro-sink.port = 4545

注意：这里的sink使用了avro，接口服务器的flume会通过rpc的方式将日志数据发给汇总日志的服务器；

4）启动：

nohup flume-ng agent -c /usr/local/apache-flume-1.7.0-bin/conf -f /usr/local/apache-flume-1.7.0-bin/conf/test-tomcat-log.conf  -n agent1 >/dev/null 2>&1 &

2、在汇总日志服务器上flume配置：

1）安装、解压、配置flume：

2）flume配置文件：

collector1.sources = AvroIn
    collector1.channels = mc1
    collector1.sinks = LocalOut
    
    collector1.sources.AvroIn.type = avro
    collector1.sources.AvroIn.bind = 10.153.137.211
    collector1.sources.AvroIn.port = 4545
    collector1.sources.AvroIn.channels = mc1
    
    collector1.channels.mc1.type = memory
    collector1.channels.mc1.capacity = 100
    collector1.channels.mc1.transactionCapacity = 100
    
    collector1.sinks.LocalOut.type = file_roll
    collector1.sinks.LocalOut.sink.directory = /data/tomcat_log_bak
    collector1.sinks.LocalOut.sink.rollInterval = 0
    collector1.sinks.LocalOut.channel = mc1

说明：

A、这里的source使用的是avro，和api接口的flume进行对接；

B、这里使用file\_roll的sink，将日志数据保存到本地磁盘；

注：bind只能写本机ip或者机器名，不能写localhost等。

3）启动：

nohup flume-ng agent -c /usr/local/apache-flume-1.7.0-bin/conf -f /usr/local/apache-flume-1.7.0-bin/conf/tomcat_collection.conf -n collector1 -Dflume.root.logger=INFO,console >/dev/null 2>&1 &

这是，我们会发现/data/tomcat\_log\_bak 目录下会生成从两台接口服务器上收集回来的日志。

二、场景二描述：

线上api接口服务通过log4j往本地磁盘上打印日志，在接口服务器上安装flume，通过exec source收集日志，然后通过avro sink将日志发送到汇总服务器上的flume；在汇总服务器上的flume，通过avro source接收到日志，然后通过hdfs sink备份到hdfs上。

![Center 1][]

假设有api接口服务器两台 10.153.140.250和10.153.140.251，汇总日志的服务器一台 10.153.137.211

1、api接口服务器上flume配置：

同上；

2、汇总服务器上flume配置：

1）安装、解压flume：

2）flume配置文件：

agent1.channels = ch1
    agent1.sources = s1
    agent1.sinks = log-sink1
    
    agent1.sources.s1.type = avro
    agent1.sources.s1.bind = 10.153.135.113
    agent1.sources.s1.port = 41414
    agent1.sources.s1.threads = 5
    agent1.sources.s1.channels = ch1
    
    agent1.channels.ch1.type = memory
    agent1.channels.ch1.capacity = 100000
    agent1.channels.ch1.transactionCapacity = 100000
    agent1.channels.ch1.keep-alive = 30
    
    agent1.sinks.log-sink1.type = hdfs
    agent1.sinks.log-sink1.hdfs.path = hdfs://hadoop-jy-namenode/data/abctt/flume
    agent1.sinks.log-sink1.hdfs.writeFormat = Text
    agent1.sinks.log-sink1.hdfs.fileType = DataStream
    agent1.sinks.log-sink1.hdfs.rollInterval = 0
    agent1.sinks.log-sink1.hdfs.rollSize = 60554432
    agent1.sinks.log-sink1.hdfs.rollCount = 0
    agent1.sinks.log-sink1.hdfs.batchSize = 1000
    agent1.sinks.log-sink1.hdfs.txnEventMax = 1000
    agent1.sinks.log-sink1.hdfs.callTimeout = 60000
    agent1.sinks.log-sink1.hdfs.appendTimeout = 60000
    agent1.sinks.log-sink1.channel = ch1

说明：

A、这里的source使用的是avro，和api接口的flume进行对接；

B、这里的sink使用的是hdfs，可以将数据写入到hdfs上，这里需要指定hadoop集群的namenode地址。（hdfs://hadoop-jy-namenode/）

3）启动：

nohup flume-ng agent -c /usr/local/apache-flume-1.7.0-bin/conf -f /usr/local/apache-flume-1.7.0-bin/conf/hdfs.conf  -n agent1 >/dev/null 2>&1 &

这时，我们会在hdfs的/data/abctt/flume目录下生成从两台接口服务器上收集回来的日志。

假设有api接口服务器两台 10.153.140.250和10.153.140.251，我们可以在接口服务器上部署flume ，将

汇总日志的服务器一台 10.153.137.211

[Center]: /images/20210511/1288e300a9124dc8b935a3c7fdfd2650.png
[Center 1]: /images/20210511/17680b2cf0ea42968aeac8fdd0d22093.png