hadoop - 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

我认为您正在这样做错误!!! 为什么？

看到您的来源是“不可拆分” ZIP。您可以逐条记录地逐一读取它们，如果不解压缩，则会得到GZIPInputStream，您将在水槽源中获取它。

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

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

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