Flume, um gz-Dateien zu streamen

Question

Ich habe einen Ordner enthält eine Menge von Gzip-Dateien. Jede gzip-Datei enthält eine XML-Datei. Ich hatte Kanal verwendet, um die Dateien in HDFS zu streamen. Im Folgenden finden Sie meine Konfigurationsdatei:

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 

Nachdem ich die Dateien in HDFS streamen, ich benutze Funken, es zu lesen mit dem folgenden Code:

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

Aber ich Problem habe, es zu lesen. Wenn ich manuell eine Gzip-Datei in den HDFS-Ordner hochlade und den obigen Spark-Code erneut ausführe, kann er sie ohne Probleme lesen. Ich bin nicht sicher, ob es wegen Gerinne ist.

Ich habe versucht, die Datei per Stream heruntergeladen und entpacken, wenn ich den Inhalt sah, zeigt es nicht mehr das XML-Format, es ist ein unlesbarer Charakter. Könnte mir jemand etwas Licht dazu bringen? Vielen Dank.

Answer 1

Ich denke du machst es Falsch !!! Warum?

Sie haben eine Quelle, die ist "Non Split-fähig" ZIP. Sie können sie teilweise als Datensatz für Datensatz lesen, wenn Sie sie nicht dekomprimieren, erhalten Sie einen GZIPInputStream, den Sie in der Kanalquelle erhalten.

Und nach dem Lesen dieses GZIP-Eingabestreams als Eingabedatensätze speichern Sie bereits gezippte Streams in einem anderen GZIP-Stream, während Sie den Senke-Typ als komprimiert ausgewählt haben.

So haben Sie in einem Gzip in HDFS gepackten Streams. :)

Ich schlage vor, ein Skript in Cron, um eine Kopie von lokalen zu HDFS tun, wird Ihr Problem lösen.