我有如下Flume代理
agent1.sources = Weather
agent1.sources.Weather.type = spooldir
agent1.sources.Weather.spoolDir = /Weather/Docs
agent1.sources.Weather.channels = MemChannel
agent1.channels = MemChannel
agent1.channels.MemChannel.type = memory
agent1.channels.MemChannel.capacity = 10000
agent1.channels.MemChannel.transactionCapacity = 1000
agent1.channels.MemChannel.deletePolicy = immediate
agent1.sinks = HDFS
agent1.sinks.HDFS.channel = MemChannel
agent1.sinks.HDFS.type = hdfs
agent1.sinks.HDFS.hdfs.path = hdfs://localhost:8020/user/flume/input/
agent1.sinks.HDFS.hdfs.fileType = DataStream
agent1.sinks.HDFS.hdfs.writeFormat = Text
agent1.sinks.HDFS.hdfs.batchSize = 1000
agent1.sinks.HDFS.hdfs.rollSize = 0
agent1.sinks.HDFS.hdfs.rollCount = 10000spool目录中的文件将自动重命名为.completed。在flume代理将该文件写入hdfs之后，文件应重命名为.completed，但在我的示例中，在代理运行之前，其重命名文件应重命名为.completed。它还将文件重命名为.completed，即使我只是手动将文件复制到假脱机目录。
另一个问题是deletepolicy即使在将文件复制到hdfs之后也不删除文件。
代理将假脱机目录文件随机写入hdfs。
它还在hdfs中创建大量tmp文件。
我是在写代理的时候做错了什么，还是在代理的时候漏掉了什么。。？？
请帮我解决这个问题。
提前谢谢
1条答案
按热度按时间iecba09b1#
我怀疑，Flume的行为是按设计，你可能会对事情的顺序感到困惑。你可以看看日志来证明这一点。
Flume是一个队列。spooldir源将从文件中读取行并将它们放入队列，然后删除或重命名文件。在该操作序列中，没有任何操作等待将事件写入接收器。
正如我所说，您可以阻止hdfs，您将在日志中看到这种行为。这个
queueSize在MemChannel将增加，直到您重新打开hdfs。