flume组件之hdfs sink

简介

flume的三大组件 source\channel\sink对应着采集位置类型\缓存类型\下沉地类型

本文主要讲解sink中的hdfs sink的常见属性以及常见问题

常用属性

type:指定sink的类型,那肯定是hdfs

hdfs.path:指定采集文件到hdfs后的路径

hdfs.filePrefix:指定在hdfs上生成文件后的前缀名

hdfs.fileSuffix:指定在hdfs上生成文件后的后缀名

hdfs.round:是否打开时间上的舍弃(就是在固定的时间内产生一个文件夹)

hdfs.roundValue:设置时间舍弃的间隔时间,默认为1

hdfs.roundUnit:设置时间舍弃的单位,默认为秒(注意跟上面的两个属性必须一块使用)

hdfs.rollInterval:设置多长时间在hdfs上产生一个新的文件,默认是30秒

hdfs.rollSize:设置文件大小等于多大时候开始滚动产生新的文件,默认是1024个字节

hdfs.rollCount:设置文件收到多少个event后开始滚动新的文件,默认是10个

举例

采集服务器/root/use.txt文件到hdfs的/flume/test下

操作:

vim exec_hdfs.conf //编辑一个新的文件在flume的conf路径下

//设置组件名
a1.sources = r1
a1.channels = c1
a1.sinks = k1//设置source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/user.txt//设置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000//设置sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/test/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute//连接
a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

常见问题

假设我们只设置了滚动条件按文件的大小滚动,只有当文件大小等于128M的时候才开始滚动新文件,那么我们在配置hdfs sink的滚动条件时候如下:

hdfs.rollInterval = 0

hdfs.rollSize = 128*1024*1024

hdfs.rollCount = 0

这个时候我们会遇到一些问题

问题1:如果文件大小一直到达不了128M的时候,文件在hdfs上会一直以临时文件的形式存在

解决方案: ①停掉flume(虽然可以解决,但是一定不要用)

②使用hdfs sink的idleTimeOut属性(例如hdfs.idleTimeOut = 30,单位为秒),监听文件的状态,如果在设置的时间内没有收到新的数据,将会把临时文件修改为最终文件

切记,这个时间没有固定的值,但是需要大于数据采集的频率,假如数据10秒钟才发送一条,那么你设置为5秒钟,那么会造成一直产生新文件

问题2:由于hdfs的副本机制,在向hdfs上上传文件的时候 ,在最后一个副本接收完成后才会认为一条数据刚传递完(不懂的同学去了解一下hdfs的上传文件的流程),这个过程是需要一定的时间的,假如第一台datanode刚好已经接收了128M的文件,但是第三台datanode还没有接收完毕,这个时候hdfs sink还没有接收到成功反馈,会继续发送,最后造成文件的大小大于128M

解决方案:让flume在检测是否上传成功的时候,只检测一个副本接收成功即可