我配置了一个flume代理，该代理从FTP服务器读取并将文件发送到hdfs接收器。
我的大问题是，我想用原始文件名将文件存储在hdfs中。我尝试使用Spooldir源，它可以正常工作，并能够使用其基名将文件存储在hdfs中，但是Flume Agent Crush:
1)如果将文件放入假脱机目录后进行写入，Flume将在其日志文件中打印错误并停止处理。
2)如果以后重用文件名，Flume将在其日志文件中打印错误并停止处理。
实际上，spooldir-source不适合我的用例。
因此，有一个想法如何使ftp源保留文件名，随后，hdfs根据文件名分别存储文件。
这是我的经纪人:
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 = pwd
agent.sources.r1.port = 22
agent.sources.r1.knownHosts = ~/.ssh/known_hosts
agent.sources.r1.work.dir = /DATA/flume_ftp_source
agent.sources.r1.fileHeader = true
agent.sources.r1.basenameHeader = true
agent.sources.r1.inputCharset = ISO-8859-1
agent.sources.r1.flushlines = true
#configure sink s1
agent.sinks.k.type = hdfs
agent.sinks.k.hdfs.path = hdfs://hostname:8020/user/admin/DATA/import_flume/agents/agent1/%Y/%m/%d/%H
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
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
最佳答案
我只是将解决方案推送到了水槽ftp github项目:
菲利普·KR
Is there a trick on how to fix the fact that the property %{basename} is missing ?
关于hadoop - 如何在ftp源flume代理中保留文件的原始基本名称，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43675459/