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

121 阅读 0 评论 80 点赞

我是靠谱客的博主明理玫瑰，最近开发中收集的这篇文章主要介绍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

三、注意事项及异常

idleTimeout 的设置
- 设置为0，如果flume程序突然宕机，就会导致 hdfs上的 .tmp后缀的文件无法会更改为完成的文件，造成一种假象，以为该文件正在写入。当程序重启时，就会有两个 .tmp文件。
- 如果idle Timeout有设置值m，当在m秒内没有数据写入，就会把tmp文件改为已完成。后面再有数据过来的时候重新生成.tmp文件。
- 建议：最好设置一个比较大的值，防止小文件产生，若不设置，宕机的话会有tmp文件
- 为了能快速查看到数据，可以设置该值较小，没数据进行就滚动，因为临时文件是不能被Hive查询到，但是这样会产生小文件
round 与 rollInterval 理解有误
- round、roundValue、roundUnit是基于路径path去滚动生成文件夹的，针对文件夹而言
- rollInterval、rollSize、rollCount是基于文件的条件限制滚动生成文件的，基于文件而言的
异常：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设置不同的文件后缀名