Flume 到 HDFS 接收器的内存通道

Question

我遇到了 Flume（Cloudera CDH 5.3 上的 1.5）问题：

spoolDir source -> memory channel -> HDFS sink

我正在尝试做的事情：每 5 分钟，大约 20 个文件被推送到假脱机目录（从远程存储中获取）。每个文件包含多行，每行是一个日志（在 JSON 中）。文件大小介于 10KB 和 1MB 之间。

当我启动代理时，所有文件都已成功推送到 HDFS。 1 分钟后（这是我在 flume.conf 中设置的），文件被滚动（删除 .tmp 后缀并关闭）。

但是，当在假脱机目录中找到新文件时，我收到消息：

org.apache.flume.source.SpoolDirectorySource: The channel is full, and cannot write data now. The source will try again after 250 milliseconds

在尝试了很多不同的配置后都没有成功（increasing/decreasing channel transactionCapacity 和 capacity，increasing/decreasing batchSize 等），我请求你的帮助。

这是我最新的 flume 配置：

# source definition
sebanalytics.sources.spooldir-source.type = spooldir
sebanalytics.sources.spooldir-source.spoolDir = /var/flume/in
sebanalytics.sources.spooldir-source.basenameHeader = true
sebanalytics.sources.spooldir-source.basenameHeaderKey = basename
sebanalytics.sources.spooldir-source.batchSize = 10
sebanalytics.sources.spooldir-source.deletePolicy = immediate
# Max blob size: 1.5Go
sebanalytics.sources.spooldir-source.deserializer = org.apache.flume.sink.solr.morphline.BlobDeserializer$Builder
sebanalytics.sources.spooldir-source.deserializer.maxBlobLength = 1610000000
# Attach the interceptor to the source
sebanalytics.sources.spooldir-source.interceptors = json-interceptor
sebanalytics.sources.spooldir-source.interceptors.json-interceptor.type = com.app.flume.interceptor.JsonInterceptor$Builder
# Define event's headers. basenameHeader must be the same than source.basenameHeaderKey (defaults is basename)
sebanalytics.sources.spooldir-source.interceptors.json-interceptor.basenameHeader = basename
sebanalytics.sources.spooldir-source.interceptors.json-interceptor.resourceHeader = resources
sebanalytics.sources.spooldir-source.interceptors.json-interceptor.ssidHeader = ssid

# channel definition
sebanalytics.channels.mem-channel-1.type = memory
sebanalytics.channels.mem-channel-1.capacity = 1000000
sebanalytics.channels.mem-channel-1.transactionCapacity = 10

# sink definition
sebanalytics.sinks.hdfs-sink-1.type = hdfs
sebanalytics.sinks.hdfs-sink-1.hdfs.path = hdfs://StandbyNameNode/data/in
sebanalytics.sinks.hdfs-sink-1.hdfs.filePrefix = %{resources}_%{ssid}
sebanalytics.sinks.hdfs-sink-1.hdfs.fileSuffix = .json
sebanalytics.sinks.hdfs-sink-1.hdfs.fileType = DataStream
sebanalytics.sinks.hdfs-sink-1.hdfs.writeFormat = Text
sebanalytics.sinks.hdfs-sink-1.hdfs.rollInterval = 3600
sebanalytics.sinks.hdfs-sink-1.hdfs.rollSize = 63000000
sebanalytics.sinks.hdfs-sink-1.hdfs.rollCount = 0
sebanalytics.sinks.hdfs-sink-1.hdfs.batchSize = 10
sebanalytics.sinks.hdfs-sink-1.hdfs.idleTimeout = 60

# connect source and sink to channel
sebanalytics.sources.spooldir-source.channels = mem-channel-1
sebanalytics.sinks.hdfs-sink-1.channel = mem-channel-1

Answer 1

完整通道意味着：通道无法从源接收更多事件，因为接收器消耗这些事件的速度比源慢。

增加信道容量只能解决问题。可能的解决方案：

改进接收器处的处理...如果接收器是自定义接收器（improving/avoiding 循环，使用更高效的后端 API，等等）。在这种情况下，这似乎是不可能的，因为您使用的是默认的 HDFS 接收器。
降低数据发送到源的频率。不过，我猜你不希望 to/cannot 因为你的处理要求而这样做。
添加更多并行工作的接收器。我对此不确定，但我可以想象 Flume 的设计师决定运行每个接收器都在一个单独的线程中。如果是这样，那么您可以尝试使用多个并行 HDFS 接收器。为了将数据分成几个接收器，您将必须使用 multiplexing selector different than the default replicating one.

HTH！

Flume 到 HDFS 接收器的内存通道

Flume memory chanel to HDFS sink

flume-ng

cloudera-cdh