Flume 流式传输 gz 文件

Question

我有一个文件夹包含很多 gzip 文件。每个 gzip 文件都包含 xml 文件。我曾使用水槽将文件流式传输到 HDFS。下面是我的配置文件：

agent1.sources = src
agent1.channels = ch
agent1.sinks = sink

agent1.sources.src.type = spooldir
agent1.sources.src.spoolDir = /home/tester/datafiles
agent1.sources.src.channels = ch
agent1.sources.src.deserializer = org.apache.flume.sink.solr.morphline.BlobDeserializer$Builder

agent1.channels.ch.type = memory
agent1.channels.ch.capacity = 1000
agent1.channels.ch.transactionCapacity = 1000

agent1.sinks.sink.type = hdfs
agent1.sinks.sink.channel = ch
agent1.sinks.sink.hdfs.path = /user/tester/datafiles
agent1.sinks.sink.hdfs.fileType = CompressedStream
agent1.sinks.sink.hdfs.codeC = gzip
agent1.sinks.sink.hdfs.fileSuffix = .gz
agent1.sinks.sink.hdfs.rollInterval = 0
agent1.sinks.sink.hdfs.rollSize = 122000000
agent1.sinks.sink.hdfs.rollCount = 0
agent1.sinks.sink.hdfs.idleTimeout = 1
agent1.sinks.sink.hdfs.batchSize = 1000

将文件流式传输到 HDFS 后，我使用 Spark 使用以下代码读取它：

df = sparkSession.read.format('com.databricks.spark.xml').options(rowTag='Panel', compression='gzip').load('/user/tester/datafiles')

但我在阅读时遇到问题。如果我手动将一个 gzip 文件上传到 HDFS 文件夹并重新运行上述 Spark 代码，它可以毫无问题地读取它。我不确定是不是因为水槽。

我尝试下载flume流式传输的文件并解压缩，当我查看内容时，它不再显示xml格式，它是一些不可读的字符。任何人都可以让我了解一下吗？谢谢。

Answer 1

我认为你这样做错了！！！为什么？

看到您的来源是 "Non Split-able" ZIP 。您可以逐条记录地读取它们，如果您不解压缩，您将获得一个 GZIPInputStream，您将在水槽源中获得它。

在将该 GZIP 输入流作为输入记录读取后，您正在将已压缩的流保存到另一个 GZIP 流中，因为您选择了已压缩的接收器类型。

因此，您已在 HDFS 的 Gzip 中压缩流式传输。 :)

我建议在 cron 中安排一个脚本从本地复制到 HDFS 将解决您的问题。