安装

1. 下载

下载地址一
下载地址二
下载方式三：

wget http://archive.apache.org/dist/flume/stable/apache-flume-1.9.0-bin.tar.gz

这里使用最新的 apache-flume-1.9.0版本

2. 解压安装

   tar zxvf apache-flume-1.9.0-bin.tar.gz -C /opt/pkg/
   

3. 改目录名他人
   mv apache-flume-1.9.0-bin/ flume

4. 配置环境变量，并让环境变量生效

   vim /etc/profile.d/env.sh
   

   # FLUME 1.9.0
   export FLUME_HOME=/opt/pkg/flume
   export PATH=$FLUME_HOME/bin:$PATH
   

   source /etc/profile.d/env.sh
   

5. 将hadoop-2.7.3安装路径下的依赖的jar软链接到flume-1.9.0/lib下：

   ln -s /user/hadoop/hadoop-2.7.3/share/hadoop/common/hadoop-common-2.7.3.jar $FLUME_HOME/lib/
   ln -s /user/hadoop/hadoop-2.7.3/share/hadoop/common/lib/commons-configuration-1.6.jar $FLUME_HOME/lib/
   ln -s /user/hadoop/hadoop-2.7.3/share/hadoop/common/lib/hadoop-auth-2.7.3.jar $FLUME_HOME/lib/
   ln -s /user/hadoop/hadoop-2.7.3/share/hadoop/common/lib/htrace-core-3.1.0-incubating.jar $FLUME_HOME/lib/
   ln -s /user/hadoop/hadoop-2.7.3/share/hadoop/common/lib/commons-io-2.4.jar $FLUME_HOME/lib/
   ln -s /user/hadoop/hadoop-2.7.3/share/hadoop/hdfs/hadoop-hdfs-2.7.3.jar $FLUME_HOME/lib/
   
   

测试

1. 验证

   bin/flume-ng version 
   
   flume 1.9.0
   Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
   Revision: d4fcab4f501d41597bc616921329a4339f73585e
   Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018
   From source with checksum 35db629a3bda49d23e9b3690c80737f9
   

2. 配置Flume HDFS Sink:
   在flume的conf目录新建一个log2hdfs.conf
   添加如下内容：

   [hadoop@master conf]$ vi log2hdfs.conf
   
   # define the agent
   a1.sources=r1
   a1.channels=c1
   a1.sinks=k1
   
   # define the source
   #上传目录类型
   a1.sources.r1.type=spooldir
   a1.sources.r1.spoolDir=/tmp/flume-logs
   #定义自滚动日志完成后的后缀名
   a1.sources.r1.fileSuffix=.FINISHED
   #根据每行文本内容的大小自定义最大长度4096=4k
   a1.sources.r1.deserializer.maxLineLength=4096
   
   # define the sink
   a1.sinks.k1.type = hdfs
   #上传的文件保存在hdfs的/flume/logs目录下
   a1.sinks.k1.hdfs.path = hdfs://master:8020/flume/logs/%y-%m-%d/
   a1.sinks.k1.hdfs.filePrefix=access_log
   a1.sinks.k1.hdfs.fileSufix=.log
   a1.sinks.k1.hdfs.batchSize=1000
   a1.sinks.k1.hdfs.fileType = DataStream
   a1.sinks.k1.hdfs.writeFormat= Text
   # roll 滚动规则:按照数据块128M大小来控制文件的写入，与滚动相关其他的都设置成0
   #为了演示，这里设置成500k写入一次
   a1.sinks.k1.hdfs.rollSize= 512000
   a1.sinks.k1.hdfs.rollCount=0
   a1.sinks.k1.hdfs.rollInteval=0
   #控制生成目录的规则:一般是一天或者一周或者一个月一次，这里为了演示设置10秒 
   a1.sinks.k1.hdfs.round=true
   a1.sinks.k1.hdfs.roundValue=10
   a1.sinks.k1.hdfs.roundUnit= second
   #是否使用本地时间
   a1.sinks.k1.hdfs.useLocalTimeStamp=true
   
   #define the channel
   a1.channels.c1.type = memory
   #自定义event的条数
   a1.channels.c1.capacity = 500000
   #flume事务控制所需要的缓存容量1000条event
   a1.channels.c1.transactionCapacity = 1000
   
   #source channel sink cooperation
   a1.sources.r1.channels = c1
   a1.sinks.k1.channel = c1
   

   注意：

   • a1.sources.r1.spoolDir目录如果不存在需要先创建
   • a1.sinks.k1.hdfs.path目录会自动创建
     • 为了演示的效果明显，这里的路径是hdfs://hadoop100:9000/flume/logs/yy-mm-dd/HH/MM/SS
     • 也就是每秒的数据都会产生滚动日志，也就是1,2条记录就会单独生成一个128M的文件。
     • 实际应该是按天或者按周、按月来生成滚动日志。

3. 修改conf/flume-env.sh，配置JDK路径（该文件事先是不存在的，需要复制一份）
   复制：

   [root@master conf]# cp flume-env.sh.template flume-env.sh
   

   编辑文件，并设置如下内容：

   #设置JAVA_HOME:
   export JAVA_HOME=/opt/pkg/jdk1.8.0_211		//等号左右不能有空格         
   #修改默认的内存：  
   export JAVA_OPTS="-Xms1024m -Xmx1024m -Xss256k -Xmn2g -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:-UseGCOverheadLimit"   
   

4. 启动flume

   • 准备

     创建/tmp/flume-logs , 并分配权限

     sudo mkdir -p  /tmp/flume-logs
     sudo chown hadoop:hadoop /tmp/flume-logs
     

   • 启动
     执行如下命令进行启动：

     [root@master flume]# bin/flume-ng agent --conf ./conf/ -f ./conf/log2hdfs.conf --name a1 -Dflume.root.logger=INFO,console
     

     

   • 测试

     # 在tmp下新建一个会话并连接。
      touch access_log123.log
     

     

   • 到Hadoop的控制台http://master:50070/查看 hdfs:master:9000/flume/logs/下有没有数据生成:

     

脚本

1. 编写Linux脚本rollingLog.sh，实现/var/log/nginx/access.log日志自动滚动到flumeLogs目录下

   • 在/opt/bin目录下新建rollinglog.sh，并添加如下内容：

     	#!/bin/bash
     	#定义日期格式
     	dataformat=`date +%Y-%m-%d-%H-%M-%S`
     	
     	#复制access.log并重命名
     	cp /var/log/nginx/access.log /var/log/nginx/access_$dataformat.log
     	
     	host=`hostname`
     	sed -i 's/^/'${host}',&/g' /var/log/nginx/access_$dataformat.log
     	#统计日志文件行数
     	lines=`wc -l < /var/log/nginx/access_$dataformat.log`
     	
     	#将格式化的日志移动到flumeLogs目录下
     	mv /var/log/nginx/access_$dataformat.log /var/log/nginx/flume-logs
     	
     	#清空access.log的内容
     	sed -i '1,'${lines}'d' /var/log/nginx/access.log
     	
     	#重启nginx , 否则 log can not roll.
     	kill -USR1 `cat /training/nginx/logs/nginx.pid`
     	
     	##返回给服务器信息
     	ls -al /var/log/nginx/flume-logs/
     

2. 编写启动Flume脚本 flume_start.sh，启动Flume

   #!/bin/bash
   /opt/pkg/flume-1.9.0/bin/flume-ng agent --conf ./conf/ -f ./conf/log2hdfs.conf --name a1 -Dflume.root.logger=INFO,console
   

3. 编写停止Flume脚本 flume_stop.sh，停止Flume

   #!/bin/bash
   
   JAR="flume"
   
   #停止flume函数
   echo "begin to stop flume process.."
   num=`ps -ef|grep java|grep $JAR|wc -l`
   echo "当前已经启动的flume进程数：$num"
   if [ "$num" != "0" ];then
   #正常停止flume
   ps -ef|grep java|grep $JAR|awk '{print $2;}'|xargs kill -9
   echo "进程已经关闭..."
   else
   echo "服务未启动，无须停止..."
   fi
   

4. 编写重启Flume脚本 flume_to_hdfs.sh，综合了前两个脚本

   #!/bin/bash
   
   #先停止正在启动的flume
   ./flume_stop.sh
   
   nohup ./flume_start.sh > nohup_output.log 2>&1 &
   echo "启动flume成功……"