flume 配置 - spoolDir 和 file roll sink 的变量

Question

我搜索过，有多个帖子介绍了如何将 'fileHeader' 和 'basenameHeader' 用于 spoolDirectory，但它似乎对我的情况不起作用。

我的目标是使用 Flume 使用 file_roll 接收器将本地目录中的文件提取到 blob 目录中，并保留文件名或至少将文件放在与文件名同名的目录。例如，

/input/january.txt -> /blob_output/january.[timestamp] or /input/january.txt -> /blob_output/january/[timestamp].txt

问题是我似乎无法调用文件名或基名的变量，因为它总是 returns 什么都没有（或空），导致输出名称中只有时间戳。这是我的 flume 配置：

training_agent.sources = src1
training_agent.channels = ch1
training_agent.sinks = sink1

training_agent.sources = src1
training_agent.channels = ch1
training_agent.sinks = sink1

training_agent.sources.src1.type = spooldir
training_agent.sources.src1.channels = ch1
training_agent.sources.src1.spoolDir = /home/training/input/
training_agent.sources.src1.batchSize = 5
training_agent.sources.src1.fileHeader = true

training_agent.channels.ch1.type = memory
training_agent.channels.capacity = 500
training_agent.channels.ch1.transactionCapacity = 50

training_agent.sinks.sink1.type = file_roll
training_agent.sinks.sink1.sink.rollInterval = 0
training_agent.sinks.sink1.channel = ch1
training_agent.sinks.sink1.sink.directory = /blobtraining_path/destination
training_agent.sinks.sink1.sink.pathManager.prefix = ${file}

Answer 1

在滚动文件接收器的documentation中，没有指定输出文件文件名的选项。

我查看了 source 以找到解决此问题的方法，但没有简单的方法可以做到。 Flume 仅使用当前时间戳生成文件名。您只能为输出文件指定前缀和扩展名。

但是，您可以扩展 DefaultPathManager 并添加执行此操作的能力，或者您可以创建一个新的自定义接收器（您自己的滚动文件接收器），它能够获取文件名 header来自 flume 事件 headers 并使用此信息创建与输入文件具有相同值的输出文件。

要创建自定义接收器，您可以查看 Flume Developer Guide 的接收器部分。

补充信息：

training_agent.sinks.sink1.sink.pathManager.prefix = ${file}

您不能直接在 flume 配置文件中使用环境变量。如果你想在你的 flume 配置中使用环境变量，你必须将它添加到你的执行命令中。

-DpropertiesImplementation=org.apache.flume.node.EnvVarResolverProperties

flume 配置 - spoolDir 和 file roll sink 的变量

flume configuration - spoolDir and file roll sink's variable

flume

flume-ng