一、需求说明

flume监控linux上一个目录(/home/flume_data)下进入的文件，并写入hdfs的相应目录下(hdfs://master:9000/flume/spool/%Y%m%d%H%M)

二、新建配置文件

1、在conf下新建配置文件hdfs-logger.conf

# Name the components on this agent
spool-hdfs-agent.sources = spool-source
spool-hdfs-agent.sinks = hdfs-sink
spool-hdfs-agent.channels = memory-channel# Describe/configure the source
spool-hdfs-agent.sources.spool-source.type = spooldir
spool-hdfs-agent.sources.spool-source.spoolDir = /home/flume_data# Describe the sink
spool-hdfs-agent.sinks.hdfs-sink.type = hdfs
spool-hdfs-agent.sinks.hdfs-sink.hdfs.path = hdfs://master:9000/flume/spool/%Y%m%d%H%M
spool-hdfs-agent.sinks.hdfs-sink.hdfs.useLocalTimeStamp = true
spool-hdfs-agent.sinks.hdfs-sink.hdfs.fileType = CompressedStream
spool-hdfs-agent.sinks.hdfs-sink.hdfs.writeFormat = Text
spool-hdfs-agent.sinks.hdfs-sink.hdfs.codeC = gzip
spool-hdfs-agent.sinks.hdfs-sink.hdfs.filePrefix = wsk
spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollInterval = 30
spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollSize = 1024
spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollCount = 0# Use a channel which buffers events in memory
spool-hdfs-agent.channels.memory-channel.type = memory
spool-hdfs-agent.channels.memory-channel.capacity = 1000
spool-hdfs-agent.channels.memory-channel.transactionCapacity = 100# Bind the source and sink to the channel
spool-hdfs-agent.sources.spool-source.channels = memory-channel
spool-hdfs-agent.sinks.hdfs-sink.channel = memory-channel

2、说明

(1)spool-hdfs-agent为agent的名字，需要在启动flume命令中的- name中配置的；

(2)/home/flume_data为flume监控采集目录；

(3)hdfs://master:9000/flume/spool/%Y%m%d%H%M，为flume输出hdfs的目录地址，%Y%m%d%H%M是输出文件夹时间格式；

(4)flume有三种滚动方式。
按照时间:spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollInterval = 30
按照大小:spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollSize = 1024
按照count:spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollCount = 0

滚动的意思是当flume监控的目录达到了配置信息中的某一条滚动方式的时候，会触发flume提交一个文件到hdfs中（即在hdfs中生成一个文件）

rollInterval默认值：30
hdfs sink间隔多长将临时文件滚动成最终目标文件，单位：秒；
如果设置成0，则表示不根据时间来滚动文件；
注：滚动（roll）指的是，hdfs sink将临时文件重命名成最终目标文件，并新打开一个临时文件来写入数据；rollSize默认值：1024
当临时文件达到该大小（单位：bytes）时，滚动成目标文件；
如果设置成0，则表示不根据临时文件大小来滚动文件；rollCount默认值：10
当events数据达到该数量时候，将临时文件滚动成目标文件；
如果设置成0，则表示不根据events数据来滚动文件；

(5)rollSize控制的大小是指的压缩前的，所以若hdfs文件使用了压缩，需调大rollsize的大小

(6)当文件夹下的某个文件被采集到hdfs上，会有个。complete的标志

(7)使用Spooling Directory Source采集文件数据时若该文件数据已经被采集，再对该文件做修改是会报错的停止的，其次若放进去一个已经完成采集的同名数据文件也是会报错停止的

(8)写HDFS数据可按照时间分区，注意改时间刻度内无数据则不会生成该时间文件夹

(9)生成的文件名称默认是前缀+时间戳，这个是可以更改的。

三、启动flume

1、命令

[root@master bin]# ./flume-ng agent --conf conf --conf-file ../conf/hdfs-logger.conf --name spool-hdfs-agent -Dflume.root.logger=INFO,console

2、向采集目录发送文件

[root@master flumeData]# cp teacher /home/flume_data/
[root@master flumeData]# cp student /home/flume_data/

[root@master flumeData]# cat teacher
chenlaoshi
malaoshi
haolaoshi
weilaoshi
hualaoshi
[root@master flumeData]# cat student
zhangsan
lisi
wangwu
xiedajiao
xieguangkun

3、控制台日志打印

20/04/21 10:08:56 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: spool-source started
20/04/21 10:09:07 INFO avro.ReliableSpoolingFileEventReader: Last read took us just up to a file boundary. Rolling to the next file, if there is one.
20/04/21 10:09:07 INFO avro.ReliableSpoolingFileEventReader: Preparing to move file /home/flume_data/teacher to /home/flume_data/teacher.COMPLETED
20/04/21 10:09:07 INFO hdfs.HDFSCompressedDataStream: Serializer = TEXT, UseRawLocalFileSystem = false
20/04/21 10:09:07 INFO hdfs.BucketWriter: Creating hdfs://master:9000/flume/spool/202004211009/wsk.1587434947074.gz.tmp
20/04/21 10:09:08 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
20/04/21 10:09:08 INFO compress.CodecPool: Got brand-new compressor [.gz]
20/04/21 10:09:17 INFO avro.ReliableSpoolingFileEventReader: Last read took us just up to a file boundary. Rolling to the next file, if there is one.
20/04/21 10:09:17 INFO avro.ReliableSpoolingFileEventReader: Preparing to move file /home/flume_data/student to /home/flume_data/student.COMPLETED

4、监控目录

[root@master flume_data]# ls
student.COMPLETED  teacher.COMPLETED

5、hdfs存储效果

下载解压打开

四、此方式的缺点

1、虽然能监控一个文件夹，但是无法监控递归的文件夹中的数据；

2、若采集时Flume挂了，无法保证重启时还从之前文件读取的那一行继续采集数据；

Flume使用Spooling Directory Source采集文件夹数据到hdfs相关推荐

flume学习（十一）：如何使用Spooling Directory Source
最近在弄一个信令数据汇聚的事情,主要目的是把FTP上的信令数据汇聚到HDFS上去存储. 逻辑是这样的:把FTP服务器上的文件下载到一台主机上,然后SCP到另外一台主机上的Spooling Direct ...
Spooling Directory Source 使用
在使用exec来监听数据源虽然实时性较高,但是可靠性较差,当source程序运行异常或者linux命令中断,都会造成数据的丢失,再恢复正常运行之前,数据的完整性无法得到保证. Spooling Dir ...
Flume 以twitter为source，kafka为channel，hdfs为sink，再用spark streaming 读kafka topic
Flume 以twitter为source,kafka为channel,hdfs为sink,再用spark streaming 读kafka topic Flume的配置文件: kafka_twitt ...
fox pro删除单条数据_Mac文件夹数据同步工具——Sync Folders Pro
Mac版同步文件夹Pro(文件夹数据同步工具)分享给大家!Mac版同步文件夹Pro是一种功能强大的文件夹数据同步工具,可帮助您同步两一个文件夹的内容,包括任何子文件夹.使用文件夹同步软件,允许您在任一 ...
不要轻易放弃丢失的U盘文件夹数据，这里有按文件夹恢复数据的技巧
U盘,全名叫USB闪存盘,是一种便携式的存储设备,是一种可以插入到电脑等电子设备上进行数据传输和存储的硬件设备.U盘的使用方便.速度高.存储容量大.稳定性高,因此被广泛用于数据备份.文档传输.音频视频 ...
C#中使用Directory实现对文件夹的常用操作
场景 Directory 命名空间:System.IO 方法 CreateDirectory(String) 在指定路径中创建所有目录和子目录,除非它们已经存在. CreateDirectory(S ...
怎样在hdfs上创建多级目录文件夹_【HDFS API编程】第一个应用程序的开发-创建文件夹...
/** * 使用Java API操作HDFS文件系统 * 关键点: * 1)创建 Configuration * 2)获取 FileSystem * 3)...剩下的就是 HDFS API的操作了 * ...
spark读取文件夹数据
1 背景数据都放在文件夹下文件夹下面每一个文件格式都一样,只是分开放了 2 步骤和读单个文件一样,只不过是把输入路径切换成文件夹就行 G:\\flow-poc\\input\\wangzish ...
python导入文件夹数据有改动_python办公自动化--批量修改文件/文件夹名称
导语今天我们来看下如何批量修改名称.这个需求在工作中比较常见的,日常生活中可能也有此类需求,比如,打包下载了一部连续剧或有声读物,每个文件名却被加上了网址.网站名称,还有一些莫名其妙的符号,整得特别 ...

Flume使用Spooling Directory Source采集文件夹数据到hdfs