flume收集日志到HDFS

作者同类文章X

转自&＃xff1a;http://www.aboutyun.com/thread-7949-1-1.html

问题导读&＃xff1a;
1.什么是flume&＃xff1f;
2.如何安装flume&＃xff1f;
3.flume的配置文件与其它软件有什么不同&＃xff1f;

一、认识flume
1.flume是什么&＃xff1f;
这里简单介绍一下&＃xff0c;它是Cloudera的一个产品
2.flume是干什么的&＃xff1f;
收集日志的
3.flume如何搜集日志&＃xff1f;
我们把flume比作情报人员
&＃xff08;1&＃xff09;搜集信息
&＃xff08;2&＃xff09;获取记忆信息
&＃xff08;3&＃xff09;传递报告间谍信息
flume是怎么完成上面三件事情的&＃xff0c;三个组件&＃xff1a;
source&＃xff1a;搜集信息
channel&＃xff1a;传递信息
sink&＃xff1a;存储信息
上面有点简练&＃xff0c;详细可以参考Flume内置channel,source,sink三组件介绍
上面我们认识了&＃xff0c;flume。
下面我们来安装flume1.5

2.分别解压&＃xff1a;
下载之后&＃xff0c;我们看到下面两个包&＃xff1a;
&＃xff08;1&＃xff09;上传Linux
 
上面两个包&＃xff0c;可以下载window&＃xff0c;然后通过WinSCP,如果不会 新手指导&＃xff1a;使用 WinSCP&＃xff08;下载&＃xff09;上文件到 Linux图文教程
&＃xff08;2&＃xff09;解压包

解压apache-flume-1.5.0-bin.tar.gz&＃xff0c;解压到usr文件夹下面

1. sudo tar zxvf apache-flume-1.5.0-bin.tar.gz

 


解压apache-flume-1.5.0-src.tar.gz&＃xff0c;解压到usr文件夹下面

1. sudo tar zxvf apache-flume-1.5.0-src.tar.gz

 

(3) src里面文件内容&＃xff0c;覆盖解压后bin文件里面的内容

1. sudo cp -ri apache-flume-1.5.0-src/* apache-flume-1.5.0-bin

2.  

 

(4)重命名

1. mv apache-flume-1.5.0-bin/ flume

 

3.配置环境变量&＃xff1a;
 

配置环境变量生效

1. source /etc/environment

3.建立配置文件
这里面的配置文件还是比较特别的&＃xff0c;不同于以往我们安装的软件&＃xff0c;我们这里可以自己建立配置文件。
首先我们建立一个 example文件

1. vi example

&＃xff0c;然后把下面内容&＃xff0c;粘帖到里面就可以了&＃xff0c;注意不要有乱码&＃xff0c;有乱码的话&＃xff0c;可以直接创建一个文件&＃xff0c;然后上传。方法也有很多&＃xff0c;能解决就好。

对于下面红字部分&＃xff0c;记得创建文件夹&＃xff0c;并且注意他们的权限一致&＃xff0c;这个比较简单的&＃xff0c;就不在书写了。对于下面的配置项&＃xff0c;可以参考flume参考文档&＃xff0c;这里面的参数很详细。

agent1表示代理名称
agent1.sources&＃61;source1
agent1.sinks&＃61;sink1
agent1.channels&＃61;channel1


#配置source1
agent1.sources.source1.type&＃61;spooldir
agent1.sources.source1.spoolDir&＃61;/usr/aboutyunlog
agent1.sources.source1.channels&＃61;channel1
agent1.sources.source1.fileHeader &＃61; false

#配置sink1
agent1.sinks.sink1.type&＃61;hdfs
agent1.sinks.sink1.hdfs.path&＃61;hdfs://master:8020/aboutyunlog
agent1.sinks.sink1.hdfs.fileType&＃61;DataStream
agent1.sinks.sink1.hdfs.writeFormat&＃61;TEXT
agent1.sinks.sink1.hdfs.rollInterval&＃61;4
agent1.sinks.sink1.channel&＃61;channel1


#配置channel1
agent1.channels.channel1.type&＃61;file
agent1.channels.channel1.checkpointDir&＃61;/usr/aboutyun_tmp123
agent1.channels.channel1.dataDirs&＃61;/usr/aboutyun_tmp

 


4.启动flume

flume-ng agent -n agent1 -c conf -f usr/flume/conf/example -Dflume.root.logger&＃61;DEBUG,console

上面注意红字部分&＃xff0c;是我们自己建立的文件&＃xff0c;而对于绿色部分&＃xff0c;则是输出调试信息&＃xff0c;也可以在配置文件中配置。


5.我们启动flume之后
会看到下面信息&＃xff0c;并且信息不停的重复。这个其实是在空文件的时候&＃xff0c;监控的信息输出。
 


一旦有文件输入&＃xff0c;我们会看到下面信息。

注意&＃xff1a;这个不要关闭&＃xff0c;我们另外开启一个shell&＃xff0c;在监控文件夹中放入要上传的文件

比如我们在监控文件夹下&＃xff0c;创建一个test1文件&＃xff0c;内容如下
 


这时候flume监控shell&＃xff0c;会有相应的如下下面变化

2014-06-02 12:01:04,066 (pool-6-thread-1) [INFO - org.apache.flume.client.avro.ReliableSpoolingFileEventReader.rollCurrentFile(ReliableSpoolingFileEventReader.java:332)] Preparing to move file /usr/aboutyunlog/test1 to /usr/aboutyunlog/test1.COMPLETED
2014-06-02 12:01:04,070 (pool-6-thread-1) [ERROR - org.apache.flume.source.SpoolDirectorySource$SpoolDirectoryRunnable.run(SpoolDirectorySource.java:256)] FATAL: Spool Directory source source1: { spoolDir: /usr/aboutyunlog }: Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume to continue processing.
java.lang.IllegalStateException: File name has been re-used with different files. Spooling assumptions violated for /usr/aboutyunlog/test1.COMPLETED
at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.rollCurrentFile(ReliableSpoolingFileEventReader.java:362)
at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.retireCurrentFile(ReliableSpoolingFileEventReader.java:314)
at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(ReliableSpoolingFileEventReader.java:243)
at org.apache.flume.source.SpoolDirectorySource$SpoolDirectoryRunnable.run(SpoolDirectorySource.java:227)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)
2014-06-02 12:01:07,749 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.hdfs.HDFSDataStream.configure(HDFSDataStream.java:58)] Serializer &＃61; TEXT, UseRawLocalFileSystem &＃61; false
2014-06-02 12:01:07,803 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:261)] Creating hdfs://master:8020/aboutyunlog/FlumeData.1401681667750.tmp
2014-06-02 12:01:07,871 (hdfs-sink1-call-runner-2) [DEBUG - org.apache.flume.sink.hdfs.AbstractHDFSWriter.reflectGetNumCurrentReplicas(AbstractHDFSWriter.java:195)] Using getNumCurrentReplicas--HDFS-826
2014-06-02 12:01:07,871 (hdfs-sink1-call-runner-2) [DEBUG - org.apache.flume.sink.hdfs.AbstractHDFSWriter.reflectGetDefaultReplication(AbstractHDFSWriter.java:223)] Using FileSystem.getDefaultReplication(Path) from HADOOP-8014
2014-06-02 12:01:10,945 (Log-BackgroundWorker-channel1) [INFO - org.apache.flume.channel.file.EventQueueBackingStoreFile.beginCheckpoint(EventQueueBackingStoreFile.java:214)] Start checkpoint for /usr/aboutyun_tmp123/checkpoint, elements to sync &＃61; 3
2014-06-02 12:01:10,949 (Log-BackgroundWorker-channel1) [INFO - org.apache.flume.channel.file.EventQueueBackingStoreFile.checkpoint(EventQueueBackingStoreFile.java:239)] Updating checkpoint metadata: logWriteOrderID: 1401681430998, queueSize: 0, queueHead: 11
2014-06-02 12:01:10,952 (Log-BackgroundWorker-channel1) [INFO - org.apache.flume.channel.file.Log.writeCheckpoint(Log.java:1005)] Updated checkpoint for file: /usr/aboutyun_tmp/log-8 position: 2482 logWriteOrderID: 1401681430998
2014-06-02 12:01:10,953 (Log-BackgroundWorker-channel1) [DEBUG - org.apache.flume.channel.file.Log.removeOldLogs(Log.java:1067)] Files currently in use: [8]
2014-06-02 12:01:11,872 (hdfs-sink1-roll-timer-0) [DEBUG - org.apache.flume.sink.hdfs.BucketWriter$2.call(BucketWriter.java:303)] Rolling file (hdfs://master:8020/aboutyunlog/FlumeData.1401681667750.tmp): Roll scheduled after 4 sec elapsed.
2014-06-02 12:01:11,873 (hdfs-sink1-roll-timer-0) [INFO - org.apache.flume.sink.hdfs.BucketWriter.close(BucketWriter.java:409)] Closing hdfs://master:8020/aboutyunlog/FlumeData.1401681667750.tmp
2014-06-02 12:01:11,873 (hdfs-sink1-call-runner-7) [INFO - org.apache.flume.sink.hdfs.BucketWriter$3.call(BucketWriter.java:339)] Close tries incremented
2014-06-02 12:01:11,895 (hdfs-sink1-call-runner-8) [INFO - org.apache.flume.sink.hdfs.BucketWriter$8.call(BucketWriter.java:669)] Renaming hdfs://master:8020/aboutyunlog/FlumeData.1401681667750.tmp to hdfs://master:8020/aboutyunlog/FlumeData.1401681667750
2014-06-02 12:01:11,897 (hdfs-sink1-roll-timer-0) [INFO - org.apache.flume.sink.hdfs.HDFSEventSink$1.run(HDFSEventSink.java:402)] Writer callback called.
2014-06-02 12:01:12,423 (conf-file-poller-0) [DEBUG - org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:126)] Checking file:conf/example for changes
2014-06-02 12:01:40,953 (Log-BackgroundWorker-channel1) [DEBUG - org.apache.flume.channel.file.FlumeEventQueue.checkpoint(FlumeEventQueue.java:137)] Checkpoint not required

上传成功之后&＃xff0c;我们去hdfs上&＃xff0c;查看上传文件&＃xff1a;
 


这样我们做到了flume上传到hadoop2.2。

完毕