HDFS Flume 水槽 - 按文件滚动答案

【问题标题】：HDFS Flume sink - Roll by FileHDFS Flume 水槽 - 按文件滚动
【发布时间】：2016-10-31 02:58:33
【问题描述】：

HDFS Flume sink 是否可以在单个文件（来自 Flume 源，例如 Spooling Directory）结束时滚动，而不是在特定字节 (hdfs.rollSize)、时间 (hdfs.rollInterval) 或事件 ( hdfs.rollCount)?

Flume 可以配置成一个文件就是一个事件吗？

感谢您的意见。

【问题讨论】：

标签： flume

【解决方案1】：

关于您的第一个问题，这是不可能的，因为接收器逻辑与源逻辑断开。我的意思是，接收器只看到被放入必须由他处理的通道的事件；接收器不知道事件是关于文件的第一个事件还是最后一个事件。

当然，您可以尝试创建自己的源（或扩展现有源），以便为事件添加标头，其值表示“这是最后一个事件”。然后，另一个自定义接收器可能会根据这样的标头运行：例如，如果标头未设置，则事件不会持久化，而是存储在内存中，直到看到标头；然后所有信息都作为 bach 保存在最终后端。另一种可能性是自定义接收器将数据保留在文件中，直到看到标题为止；然后关闭文件并打开另一个文件。

关于你的第二个问题，这取决于水槽。 spooldir 源的行为基于 deserializer 参数；默认值是LINE，什么意思：

指定用于将文件解析为事件的反序列化器。默认将每一行解析为一个事件。指定的类必须实现 EventDeserializer.Builder。

但是可以配置其他自定义Java类，如上所述；例如，对整个文件进行反序列化。

【讨论】：

关于标头解决方案，我现在在想，尽管通道默认情况下表现为 FIFO 队列，但当事件无法持久时（例如，与后端）。在这种情况下，假设在队列开始时重新注入事件以进行更改重试，然后任何订单都将完全丢失。

【解决方案2】：

您可以将 rollsize 设置为一个较小的数字，结合 BlobDeserializer 以逐个文件加载而不是组合成块。当您有不可分割的二进制文件（例如 PDF 或 gz 文件）时，这非常有用。

这是相关配置的一部分：

#Set deserializer to BlobDeserializer and set the maximum blob size to be 1GB.
#Notice that the blobs have to fit in memory so this doesn't work for files that cannot fit in memory.
agent.sources.spool.deserializer = org.apache.flume.sink.solr.morphline.BlobDeserializer$Builder
agent.sources.spool.deserializer.maxBlobLength = 1000000000

#Set rollSize to 1024 to avoid combining multiple small files into one part.
agent.sinks.hdfsSink.hdfs.rollSize = 1024
agent.sinks.hdfsSink.hdfs.rollCount = 0
agent.sinks.hdfsSink.hdfs.rollInterval = 0

【讨论】：

【解决方案3】：

“能否配置 Flume 以使单个文件成为单个事件？”问题的答案是的。

您只需将以下属性配置为 1：

hdfs.rollCount = 1

我正在为您的第一个问题寻找解决方案，因为有时文件太大，需要将文件分成几块。

【讨论】：

【解决方案4】：

您可以使用hdfs.path 中的任何事件标头。 (https://flume.apache.org/FlumeUserGuide.html#hdfs-sink)

如果您使用假脱机目录源，您可以使用fileHeaderKey 或basenameHeaderKey (https://flume.apache.org/FlumeUserGuide.html#spooling-directory-source) 启用将文件名放入事件中。

Flume 可以配置成一个文件就是一个事件吗？

可以，但是不推荐。底层实现（protobuf）将文件大小限制为 64m。由于其架构和设计，Flume 事件的规模较小。（容错等）

【讨论】：