如何使 flume 将文件加载到 hdfs，hdfs 从不关闭文件 .tmp 并按名称重命名文件。

Question

实际上我有 2 个问题，我的第一个问题是：在 flume 代理刷新整个文件后如何使 HDFS 关闭文件（例如 .123456789.tmp ）。事实上，文件从未关闭，直到我强制 flume 代理停止。我相信有一种使用 4 个参数的方法如下：

hdfs.rollSize = 0 
hdfs.rollCount =0
hdfs.rollInterval = 0
hdfs.batchsize =    1000000

那么，我的第二个问题是，我的代理 flume 从 SFTP 服务器接收文件，而我需要将每个文件名保存在 hdfs 中。它适用于 spooldir 类型，但不适用于 SFTP！有什么想法吗？

我的 flume 代理配置文件如下：

agent.sources = r1 
agent.channels = c1
agent.sinks = k

configure ftp source

agent.sources.r1.type = org.keedio.flume.source.mra.source.Source
agent.sources.r1.client.source = sftp
agent.sources.r1.name.server = ip
agent.sources.r1.user = user
agent.sources.r1.password = secret
agent.sources.r1.port = 22
agent.sources.r1.knownHosts = ~/.ssh/known_hosts
agent.sources.r1.work.dir = /DATA/test/flumrFTP
agent.sources.r1.fileHeader = true
agent.sources.r1.basenameHeader = true
agent.sources.r1.inputCharset = ISO-8859-1
#agent.sources.r1.batchSize = 1000
agent.sources.r1.flushlines = true

configure sink s1
agent.sinks.k.type = hdfs
agent.sinks.k.hdfs.path =  hdfs://hostname:8000/user/admin/DATA/import_flume/
agent.sinks.k.hdfs.filePrefix = %{basename}
agent.sinks.k.hdfs.rollCount = 0
agent.sinks.k.hdfs.rollInterval = 0
agent.sinks.k.hdfs.rollSize = 0
agent.sinks.k.hdfs.useLocalTimeStamp = true
agent.sinks.k.hdfs.batchsize =    1000000
agent.sinks.k.hdfs.fileType = DataStream

Use a channel which buffers events in memory
agent.channels.c1.type = memory
agent.channels.c1.capacity =  1000000
agent.channels.c1.transactionCapacity =   1000000

agent.sources.r1.channels = c1
agent.sinks.k.channel = c1

Answer 1

尝试设置变量

hdfs.rollInterval滚动当前文件前等待的秒数

此设置会在您设置的秒数后关闭文件。我将我的设置为 200 秒，并且正在加载较小的文件

如何使 flume 将文件加载到 hdfs，hdfs 从不关闭文件 .tmp 并按名称重命名文件。

how to make flume load files to hdfs, hdfs never close file .tmp and rename file by name.

ftp

hdfs

sftp

hadoop2

flume