只有一个文件从kafka到hdfs与水槽

Question

我正在尝试通过水槽从kafka将数据放入hdfs中。kafka_producer每10秒发送一条消息。我想在hdfs上的一个文件中收集所有消息。这是我使用的水槽配置，但它在hdfs上存储了许多文件（一个用于消息）：

agent1.sources.kafka-source.type = org.apache.flume.source.kafka.KafkaSource
agent1.sources.kafka-source.zookeeperConnect = localhost:2181
agent1.sources.kafka-source.topic = prova
agent1.sources.kafka-source.groupId = flume
agent1.sources.kafka-source.channels = memory-channel
agent1.sources.kafka-source.interceptors = i1
agent1.sources.kafka-source.interceptors.i1.type = timestamp
agent1.sources.kafka-source.kafka.consumer.timeout.ms = 100
agent1.channels.memory-channel.type = memory
agent1.channels.memory-channel.capacity = 10000
agent1.channels.memory-channel.transactionCapacity = 1000
agent1.sinks.hdfs-sink.type = hdfs
agent1.sinks.hdfs-sink.hdfs.path = hdfs://localhost:9000/input
agent1.sinks.hdfs-sink.hdfs.rollInterval = 5
agent1.sinks.hdfs-sink.hdfs.rollSize = 0
agent1.sinks.hdfs-sink.hdfs.rollCount = 0
agent1.sinks.hdfs-sink.hdfs.fileType = DataStream
agent1.sinks.hdfs-sink.channel = memory-channel
agent1.sources = kafka-source
agent1.channels = memory-channel
agent1.sinks = hdfs-sink

附言我从一个文件开始.csv。kafka 生产者获取文件并选择一些感兴趣的字段，然后每 10 秒发送一个条目。Flume将条目存储在Hadoophdfs上，但存储在许多文件中（1个条目=1个文件）。我希望所有条目都在一个文件中。如何更改水槽的配置？

Answer 1

将rollInterval设置为0，因为您不希望根据时间生成不同的文件。如果您想根据数字条目s或事件进行计算，请更改rollCount值。例如，如果要在一个文件中保存10个事件或条目：

agent1.sinks.hdfs-sink.hdfs.rollInterval = 0
agent1.sinks.hdfs-sink.hdfs.rollSize = 0
agent1.sinks.hdfs-sink.hdfs.rollCount = 10

Answer 2

看来，flume目前确实被设置为在HDFS上为每个输入文件创建一个文件。

正如这里所建议的，您可以通过编写一个周期性的猪（或map减少）作业来处理这个问题，该作业获取所有输入文件并将它们组合起来。

减少文件数的另一个选项可能是减少入站文件的频率。