如何使用 Flume 将数据实时写入 HDFS？

Question

我正在使用 Flume 将传感器数据存储在 HDFS 中。一旦通过 MQTT 接收到数据。订阅者将 JSON 格式的数据发布到 Flume HTTP 监听器。它目前工作正常，但问题是水槽在我停止之前不会写入 HDFS 文件（或文件大小达到 128MB）。我正在使用 Hive 在读取时应用架构。不幸的是，生成的配置单元表仅包含 1 个条目。这是正常的，因为 Flume 没有将新的数据写入文件（由 Hive 加载）。

有什么方法可以强制 Flume 以近乎实时的方式将新数据写入 HDFS？所以，我不需要重新启动它或使用小文件？

这是我的水槽配置：

# Name the components on this agent
emsFlumeAgent.sources = http_emsFlumeAgent
emsFlumeAgent.sinks = hdfs_sink
emsFlumeAgent.channels = channel_hdfs

# Describe/configure the source
emsFlumeAgent.sources.http_emsFlumeAgent.type = http
emsFlumeAgent.sources.http_emsFlumeAgent.bind = localhost
emsFlumeAgent.sources.http_emsFlumeAgent.port = 41414

# Describe the sink
emsFlumeAgent.sinks.hdfs_sink.type = hdfs
emsFlumeAgent.sinks.hdfs_sink.hdfs.path = hdfs://localhost:9000/EMS/%{sensor}
emsFlumeAgent.sinks.hdfs_sink.hdfs.rollInterval = 0
emsFlumeAgent.sinks.hdfs_sink.hdfs.rollSize = 134217728
emsFlumeAgent.sinks.hdfs_sink.hdfs.rollCount=0

#emsFlumeAgent.sinks.hdfs_sink.hdfs.idleTimeout=20
# Use a channel which buffers events in memory
emsFlumeAgent.channels.channel_hdfs.type = memory
emsFlumeAgent.channels.channel_hdfs.capacity = 10000
emsFlumeAgent.channels.channel_hdfs.transactionCapacity = 100

# Bind the source and sinks to the channel
emsFlumeAgent.sources.http_emsFlumeAgent.channels = channel_hdfs 
emsFlumeAgent.sinks.hdfs_sink.channel = channel_hdfs

Answer 1

我认为这里的棘手之处在于您希望近乎实时地将数据写入 HDFS，但也不想要小文件（出于显而易见的原因），这可能很难实现。

您需要在以下两个参数之间找到最佳平衡：

hdfs.rollSize (Default = 1024) - 触发滚动的文件大小，以字节为单位（0：根据文件大小从不滚动）

和

hdfs.batchSize (Default = 100) - 在刷新到 HDFS 之前写入文件的事件数

如果您的数据在首选时间段内不太可能达到 128 MB，那么您可能需要减少 rollSize，但仅限于不会遇到 small files problem 的程度。

由于您没有在 HDFS 接收器中设置任何批量大小，您应该会在每 100 条记录后看到 HDFS 刷新的结果，但是一旦刷新的记录的大小共同达到 128 MB，内容将被汇总到一个128 MB 文件。 这也没有发生吗？你能确认一下吗？

希望这会有所帮助！