How to read a compressed Spark eventLog?

user_3622283 0
Viewed 1018

When I try to read a Spark 2.4.4 eventLog compressed with lz4, I obtain an empty DataFrame:

cd /opt/spark-2.4.4-bin-hadoop2.7
bin/spark-shell --master=local --conf spark.eventLog.enabled=true --conf spark.eventLog.compress=true --conf spark.io.compression.codec=lz4 --driver-memory 4G --driver-library-path=/opt/hadoop-2.7.1/lib/native/

// Trying to read an event log from a previous session
spark.read.option("compression", "lz4").json(s"file:///tmp/spark-events/local-1589202668377.lz4")

// res0: org.apache.spark.sql.DataFrame = []

However it works fine when I read an uncompressed eventLog:

bin/spark-shell --master=local --conf spark.eventLog.enabled=true --conf spark.eventLog.compress=false
spark.read.json(s"file:///tmp/spark-events/${sc.applicationId}.inprogress").printSchema

//root
// |-- App ID: string (nullable = true)
// |-- App Name: string (nullable = true)
// |-- Block Manager ID: struct (nullable = true)
// |    |-- Executor ID: string (nullable = true)

I also tried to read an eventLog compressed with snappy, same result.

lz4 apache-spark scala
2 Answers

Try doing

spark.read.json("dbfs:/tmp/compress/part-00000.lz4")
spark.conf.set("spark.io.compression.codec","org.apache.spark.io.LZ4CompressionCodec")

If it does not works there might be a good chance that your lz4 is not compatible with org.apache.hadoop.io.compress.Lz4Codec Below is the open issue link for the same lz4 incompatibility between OS and Hadoop

user_63035790

I can actually decompress the file first, using the Spark codec, and then read the decompressed file:

import java.io.{FileInputStream, FileOutputStream}
import org.apache.commons.io.IOUtils
import org.apache.spark.io.LZ4CompressionCodec

val inFile = "/tmp/spark-events/local-1589202668377.lz4"
val outFile = "/tmp/spark-events/local-1589202668377" 
val codec = new LZ4CompressionCodec(sc.getConf)
val is = codec.compressedInputStream(new FileInputStream(inFile))
val os = new FileOutputStream(outFile)
IOUtils.copyLarge(is, os)
os.close()
is.close()

spark.read.json(outFile).printSchema

It would be nicer if I could directly read the inFile without using a temporary storage, but at least I can get access to the data.

user_36222830
Related