vlambda博客
学习文章列表

Flume之HDFS Sink 的参数解析及异常处理

一、配置详解

type
Sink类型为hdfs

hdfs.path
HDFS存储路径,支持按照时间分区
集群的NameNode名字:
单节点:hdfs://主机名(ip):9000/%Y/%m/%d/%H
HA集群:hdfs://nameservice(高可用NameNode服务名称)/%Y/%m/%d/%H

hdfs.filePrefix
默认值:FlumeData
Event输出到HDFS的文件名前缀

hdfs.fileSuffix
Event输出到HDFS的文件名后缀

hdfs.inUsePrefix
临时文件的文件名前缀
Flume首先将Event输出到HDFS指定目录的临时文件中,再根据相关规则重命名为目标文件

hdfs.inUseSuffix
默认值:.tmp
临时文件名后缀

hdfs.rollInterval
默认值:30
间隔多久将临时文件滚动成最终目标文件,单位:秒
如果设置为0,则表示不根据时间滚动文件
注:滚动(roll)指的是,HDFS Sink将临时文件重命名成最终目标文件,并新打开一个临时文件来写数据

hdfs.rollSize
默认值:1024
当临时文件达到该大小时,滚动成目标文件,单位:byte
该值设置为0,则表示文件不根据文件大小滚动生成

hdfs.rollCount
默认值:10
当Event数据达到该数量时,将临时文件滚动生成目标文件
该值设置为0,则表示文件不根据Event数滚动生成

hdfs.idleTimeout
默认值:0
当目前被打开的临时文件在该参数指定的时间内,没有任何数据写入,则将该临时文件关闭并重命名成目标文件,单位:秒
该值设置为0,则表示禁用此功能,不自动关闭临时文件

hdfs.round
默认值:false
用于HDFS文件按照时间分区,时间戳向下取整

hdfs.roundValue
默认值:1
当round设置为true,配合roundUnit时间单位一起使用,例如roundUnit值为minute
该值设置为1则表示一分钟之内的数据写到一个文件中,相当于每一分钟生成一个文件

hdfs.roundUnit
默认值:second
按时间分区使用的时间单位,可以选择second(秒)、minute(分钟)、hour(小时)三种粒度的时间单位
示例:
a1.sinks.k1.hdfs.path = hdfs://nameservice/flume/events/%y/%m/%d/%H/%M
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
当时间为2022-04-05 17:38:59时候,hdfs.path依然会被解析为:/flume/events/2022/04/05/17/30
因为设置的是舍弃10分钟内的时间,因此,该目录每10分钟新生成一个

hdfs.batchSize
默认值:100
每个批次刷写到HDFS的Event数量

hdfs.codeC
默认值:不采用压缩
文件压缩格式,目前支持的压缩格式有gzip、bzip2、lzo、lzop、snappy

hdfs.fileType
默认值:SequenceFile
文件类型,包括:SequenceFile、DataStream、CompressedStream
该值设置为DataStream,则输出的文件不会进行压缩,不需要设置hdfs.codeC指定压缩格式。
该值设置为CompressedStream,则对输出的文件进行压缩,需要设置hdfs.codeC指定压缩格式

hdfs.maxOpenFiles
默认值:5000
最大允许打开的HDFS文件数,当打开的文件数达到该值,则最早打开的文件将会被关闭

hdfs.minBlockReplicas
默认值:HDFS副本数
写入HDFS文件块的最小副本数
该参数会影响文件的滚动配置,一般将该参数配置成1,才可以按照配置正确滚动文件

hdfs.writeFormat
默认值:Writable
文件的格式,目前可以选择Text或者Writable两种格式

hdfs.callTimeout
默认值:10000
操作HDFS文件的超时时间,如果需要写入HDFS文件的Event数比较大或者发生了打开、写入、刷新、关闭文件超时的问题,可以根据实际情况适当增大超时时间,单位:毫秒

hdfs.threadsPoolSize
默认值:10
每个HDFS Sink执行HDFS IO操作打开的线程数

hdfs.rollTimerPoolSize
默认值:1
HDFS Sink根据时间滚动生成文件时启动的线程数

hdfs.timeZone
默认值:Local Time本地时间
写入HDFS文件使用的时区

hdfs.useLocalTimeStamp
默认值:false
是否使用本地时间替换Event头信息中的时间戳

hdfs.closeTries
默认值:0
在发起关闭尝试后,尝试重命名临时文件的次数
如果设置为1,表示重命名一次失败后不再继续尝试重命名操作,此时待处理的文件将处于打开状态,扩展名为.tmp
如果设置为0,表示尝试重命名操作次数不受限制,直到文件最终被重命名成功
如果close调用失败,文件可能仍然会处于打开状态,但是文件中的数据将保持完整,文件会在Flume重启后关闭

hdfs.retryInterval
默认值:180 秒
连续尝试关闭文件的时间间隔。如果设置为0或小于0的数,第一次尝试关闭文件失败后将不会继续尝试关闭文件,文件将保持打开状态或者以“.tmp”扩展名结尾的临时文件
如果设置为0,表示不尝试,相当于于将hdfs.closeTries设置成1

serializer
默认值:TEXT
序列化方式,可选值有TEXT、avro_event或者实现EventSerializer.Builder接口的类

kerberosPrincipal
HDFS安全认证kerberos配置

kerberosKeytab
HDFS安全认证kerberos配置

proxyUser
代理用户

二、简单模板

agent_name.sources = source_name
agent_name.channels = channel_name
agent_name.sinks = sink_name

# source
agent_name.sources.source_name.type = avro
XXX
XXX

# channel
agent_name.channels.channel_name.type = file
XXX
XXX

# sink
agent_name.sinks.sink_name.type = hdfs
agent_name.sinks.sink_name.hdfs.path = hdfs://${HA_NameNode_Name}/flume_data/yr=%Y/mon=%m/day=%d/hr=%H
agent_name.sinks.sink_name.hdfs.writeFormat = Text
agent_name.sinks.sink_name.hdfs.fileSuffix = _${hdfsFileSuffix}.log
agent_name.sinks.sink_name.hdfs.fileType = DataStream
agent_name.sinks.sink_name.hdfs.filePrefix = %Y%m%d%H%M
agent_name.sinks.sink_name.hdfs.useLocalTimeStamp = true
agent_name.sinks.sink_name.hdfs.rollInterval = 0
agent_name.sinks.sink_name.hdfs.rollSize = 125829120
agent_name.sinks.sink_name.hdfs.rollCount = 0
agent_name.sinks.sink_name.hdfs.minBlockReplicas = 1
agent_name.sinks.sink_name.hdfs.round = true
agent_name.sinks.sink_name.hdfs.roundValue = 1
agent_name.sinks.sink_name.hdfs.roundUnit = hour
agent_name.sinks.sink_name.hdfs.idleTimeout = 600

# source | channel | sink 关联
agent_name.sources.source_name.channels = channel_name
agent_name.sinks.sink_name.channel = channel_name

三、注意事项及异常

  1. idleTimeout 的设置
    • 设置为0,如果flume程序突然宕机,就会导致 hdfs上的 .tmp后缀的文件无法会更改为完成的文件,造成一种假象,以为该文件正在写入。当程序重启时,就会有两个 .tmp文件。
    • 如果idle Timeout有设置值m,当在m秒内没有数据写入,就会把tmp文件改为已完成。后面再有数据过来的时候重新生成.tmp文件。
    • 建议:最好设置一个比较大的值,防止小文件产生,若不设置,宕机的话会有tmp文件
    • 为了能快速查看到数据,可以设置该值较小,没数据进行就滚动,因为临时文件是不能被Hive查询到,但是这样会产生小文件
  2. round 与 rollInterval 理解有误
    • round、roundValue、roundUnit是基于路径path去滚动生成文件夹的,针对文件夹而言
    • rollInterval、rollSize、rollCount是基于文件的条件限制滚动生成文件的,基于文件而言的
  3. 异常:Error while trying to hflushOrSync
    • 问题排查:通过查看不同Flume的Agent日志发现,同名的文件被不同的Flume Agent打开,在文件第二次打开后,先前打开的Agent拥有的token就失效了,因此无法关闭它,就会不断的报错:Error while trying to hflushOrSync!
    • 查看之前的flume配置文件发现,每一个Flume-Agent配置的hdfsSink是完全一样的,每个Flume-Agent读取的source相同,有很大概率会出现多个Fume-Agent同时写同名文件,导致部分Flume-Agent无法继续。
    • 解决方案:不同Flume设置不同的文件后缀名