1. 需求
使用Flume从文件夹中采集数据并上传到HDFS中。要完成这个任务就需要使用在采集数据时使用Spooling Directory Source组件;传输数据时为了保证数据没有丢失风险,使用File Channel组件;输出数据时使用HDFS Sink。
2. 配置
Flume各个组件的参数很多,因此通常复制官网的各组件样例程序并参照参数表进行修改。
2.1 Source
Source组件使用Spooling Directory Source,本次实践需要指定两个参数,分别是:
-
type
参数指定组件类型; -
spoolDir
参数指定要采集数据的源文件夹。
根据需求编写的Source配置如下:
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /data/log/studentDir
2.2 Channel
Channel组件使用File Channel,本次实践需要指定三个参数,分别是:
-
type
参数指定组件类型; -
checkpointDir
参数指定存放检查点文件的目录; -
dataDirs
表示存放数据的目录。
根据需求编写的Channel配置如下:
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data
2.3 Sink
Sink组件使用HDFS Sink,本次实践需要指定八个参数,分别是:
-
type
参数指定组件类型; -
path
参数指定最终数据要上传到的HDFS目录地址; -
filePrefix
参数指定存放数据的文件前缀; -
fileType
参数指定存放文件的类型,共有三种:SequenceFile、DataStream和CompressedStream,其中SequenceFile在Hadoop的MapReduce任务解决小文件问题时使用过; -
writeFormat
参数指定写文件的格式,有两种:Text和Writable(默认),如果后期想使用Hive或者Impala操作这份数据的话,必须在生成数据之前设置为Text,Text表示是普通文本数据; -
rollInterval
参数指定Sink间隔多长将临时文件滚动成最终目标文件,以秒为单位,默认值为30s。如果设置成0,则表示不根据时间来滚动文件(滚动roll指的是,Sink将临时文件重命名成最终目标文件,并新打开一个临时文件来写入数据); -
rollSize
参数指定当临时文件达到多大时(单位:bytes),滚动成目标文件,默认值为1024。如果设置成0,则表示不根据临时文件大小来滚动文件; -
rollCount
参数指定当events数据达到该数量时候,将临时文件滚动成目标文件,默认值为10。如果设置成0,则表示不根据events数据来滚动文件。
根据需求编写的Sink配置如下:
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://bigData01:9000/flume/studentDir
a1.sinks.k1.hdfs.filePrefix = stu-
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
2.4 完整的配置代码
将上述三个组件的配置完成后,再将其连接起来(设置Source后接的Channel和Sink前接的Channel),完整代码如下:
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /data/log/studentDir
# Describe the sink
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data
# Use a channel
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://bigData01:9000/flume/studentDir
a1.sinks.k1.hdfs.filePrefix = stu-
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
3. 实践
在运行Flume之前应该先检查建立采集数据的文件夹和文件,且系统此时直接启动会报错提示找不到SequenceFile,虽然我们已经把fileType改为了DataStream,但是Flume默认还是会加载这个类。这个问题的解决思路有两种:文章来源:https://www.toymoban.com/news/detail-511230.html
- 第一种解决思路:把SequenceFile相关的jar包都拷贝到Flume的lib目录下。但后续还会报错缺少HDFS相关的jar包,再拷贝对应的jar包即可。
- 第二种解决思路:把这个节点设置为Hadoop集群的一个客户端节点,即在这个节点上配置Hadoop的框架。
解决了这个问题后,在a1.sinks.k1.hdfs.path
路径的机器上开启Hadoop服务框架,即可启动Flume服务采集信息并上传,运行过程如下:文章来源地址https://www.toymoban.com/news/detail-511230.html
- 运行Flume
bin/flume-ng agent --name a1 --conf conf --conf-file conf/file2hdfs.conf -Dflume.root.logger=INFO,console
- 检查HDFS的上传结果
可以看到文件后有一个.tmp
的后缀,是因为当前该文件并未达到配置中设置的滚动的要求,后续的数据依旧要写入到该文件。若要去掉这个后缀那根据配置需要等待1h或者等文件达到128M,除去配置的条件外,我们也可以强制关闭这个Agent来达到目的。但需要注意一点,当我们再重启Agent后,并不会给这个文件再给加上.tmp
后缀,每次停止之前都会把所有的文件解除占用状态,下次启动的时候如果有新数据,则会产生新的文件,这其实仅仅模拟了自动切文件的效果。
[root@bigData01 soft]# hdfs dfs -ls /flume/studentDir
Found 1 items
-rw-r--r-- 1 root supergroup 65 2023-02-03 06:23 /flume/studentDir/stu-.1675376620835.tmp
[root@bigData01 soft]# hdfs dfs -cat /flume/studentDir/*
jack 18 male
jessic 20 female
tom 17 male
- 检查数据源文件
为了防止重复读取同一个文件,Flume会在读取过的文件加上COMPLETED
后缀,再不关闭Flume的前提下再加入一个class2.dat文件,会发现该文件立即读取并写入HDFS文件中。
[root@bigData02 studentDir]# ll
total 4
-rw-r--r--. 1 root root 65 Feb 3 05:31 class1.dat.COMPLETED
[root@bigData02 studentDir]# vi class2.dat
[root@bigData02 studentDir]# ll
total 8
-rw-r--r--. 1 root root 65 Feb 3 05:31 class1.dat.COMPLETED
-rw-r--r--. 1 root root 15 Feb 3 06:26 class2.dat.COMPLETED
-------------------------------------------------------------------
[root@bigData01 soft]# hdfs dfs -cat /flume/studentDir/*
jack 18 male
jessic 20 female
tom 17 male
Chaoql 23 male
到了这里,关于【Flume】Flume实践之采集文件内容上传至HDFS的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!