一、下载安装包，并上传

官网下载地址：http://flume.apache.org/download.html

或:https://download.csdn.net/download/qq_41622603/12914112

上传到/opt/software目录下，进入目录执行：tar -zxvf apache-flume-1.9.0-bin.tar.gz

这里还要修改两个地方，当然你也可以等报错的时候再回来修改

(1)flume内依赖的guava.jar和hadoop内的版本不一致会造成

ERROR [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.sink.hdfs.HDFSEventSink.process:459) - process failed java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)V at org.apache.hadoop.conf.Configuration.set(Configuration.java:1357) at org.apache.hadoop.conf.Configuration.set(Configuration.java:1338) at org.apache.hadoop.conf.Configuration.setBoolean(Configuration.java:1679) at org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:221) at org.apache.flume.sink.hdfs.BucketWriter.append(BucketWriter.java:572) at org.apache.flume.sink.hdfs.HDFSEventSink.process(HDFSEventSink.java:412) at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67) at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145) at java.lang.Thread.run(Thread.java:748)

我们需要查看hadoop安装目录下share/hadoop/common/lib内guava.jar版本 查看flume安装目录下lib内guava.jar的版本 如果两者不一致，删除版本低的，并拷贝高版本过去

(2)启动 flume的时候会出现错误：找不到org.apache.flume.tools.GetJavaProperty

需要编辑/opt/software/apache-flume-1.9.0-bin/bin目录下的flume-ng文件

在110行的位置加入：2>/dev/null | grep hbase

 

二、编辑配置文件

在/opt目录下创建conf1文件(目录和文件名没有限制)

cd /opt

vi conf1

加入下面两个配置的其中一个

用exec方式收集数据：监听文件

a1.sources = r1 a1.sinks = k1 a1.channels = c1 #读取record.list文件的内容，同时该文件内容会不断新增 a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/data/loganalysis/record.list #将Channel设置为fileChannel #配置sink的类型为hdfs a1.sinks.k1.type = hdfs #配置输出到HDFS的路径，根据日期格式分开存储文件 a1.sinks.k1.hdfs.path = hdfs://node1:8020/flume/data/%Y-%m-%d/%H%M #a1.sinks.k1.hdfs.path = hdfs://node1:8020/flume/data #生成的文件前缀 #a1.sinks.k1.hdfs.filePrefix = log #配置文件滚动时间间隔(秒),0则代表不使用该方式判断文件滚动 a1.sinks.k1.hdfs.rollInterval = 0 #配置文件滚动大小(byte) a1.sinks.k1.hdfs.rollSize = 10240000 #配置文件滚动事件数目,0则代表不使用该方式判断文件滚动 a1.sinks.k1.hdfs.rollCount = 0 #rollInterval,rollSize,rollCount其中两个要设置为0，三个判断是否生成新的文件的方式，只能使用其中一个 a1.sinks.k1.hdfs.idleTimeout= 3 a1.sinks.k1.hdfs.fileType = DataStream a1.sinks.k1.hdfs.round = true a1.sinks.k1.hdfs.roundValue = 10 a1.sinks.k1.hdfs.roundUnit = minute a1.sinks.k1.hdfs.useLocalTimeStamp = true #设置Channel类型为file a1.channels.c1.type = file #设置断点目录 a1.channels.c1.checkpointDir = /opt/software/dataCheckpointDir #设置channel数据存放目录 a1.channels.c1.dataDirs = /opt/software/dataDir #绑定Source和Sink到Channel上 a1.sinks.k1.channel = c1 a1.sources.r1.channels = c1

用spooldir方式收集数据：监听目录

a1.sources = r1 a1.sinks = k1 a1.channels = c1 #配置source使用的类型。 a1.sources.r1.type = spooldir #配置source读取文件的目录(本地目录) a1.sources.r1.spoolDir = /opt/flume #配置sink的类型为hdfs a1.sinks.k1.type = hdfs #配置输出到HDFS的路径，根据日期格式分开存储文件 a1.sinks.k1.hdfs.path = hdfs://node1:8020/flume/data/%Y-%m-%d/%H%M #a1.sinks.k1.hdfs.path = hdfs://node1:8020/flume/data #生成的文件前缀 #a1.sinks.k1.hdfs.filePrefix = log #配置文件滚动时间间隔(秒),0则代表不使用该方式判断文件滚动 a1.sinks.k1.hdfs.rollInterval = 0 #配置文件滚动大小(byte) a1.sinks.k1.hdfs.rollSize = 10240000 #配置文件滚动事件数目,0则代表不使用该方式判断文件滚动 a1.sinks.k1.hdfs.rollCount = 0 #rollInterval,rollSize,rollCount其中两个要设置为0，三个判断是否生成新的文件的方式，只能使用其中一个 a1.sinks.k1.hdfs.idleTimeout= 3 a1.sinks.k1.hdfs.fileType = DataStream a1.sinks.k1.hdfs.round = true a1.sinks.k1.hdfs.roundValue = 10 a1.sinks.k1.hdfs.roundUnit = minute a1.sinks.k1.hdfs.useLocalTimeStamp = true #设置Channel类型为Memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 #绑定Source和Sink到Channel上 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1

 

三、运行，测试：

conf:为第二步写的conf配置文件的路径

a1:conf配置文件的前缀a1

flume-ng agent --conf-file conf --name a1 -Dflume.root.logger=INFO,console

这里测试用的是监听目录的方式

当我们往/opt/flume放入文件的，flume会把文件写到hdfs