CODEWITHSUNDEEP
csvapache-sparkapache-spark-sql
G_cy
G_cy

Reputation: 1045

How does Spark SQL read compressed csv files?

I have tried with api spark.read.csv to read compressed csv file with extension bz or gzip. It worked. But in source code I don't find any option parameter that we can declare the codec type.

Even in this link, there is only setting for codec in writing side. Could anyone tell me or give the path to source code that showing how spark 2.x version deal with the compressed csv file.

Upvotes: 5

Views: 10722

Answers (2)

maxmithun
maxmithun

Reputation: 1137

You dont have to do anything special for the gz compressed csv,tsv file to get read by spark 2.x version. The below code is tried with spark 2.0.2

val options= Map("sep" -> ",")
val csvRDD = spark.read.options(options).csv("file.csv.gz")

I have done similarly for tab separated gz files

val options= Map("sep" -> "\t")
val csvRDD = spark.read.options(options).csv("file.tsv.gz")

Also you can specify the folder to read mulitple .gz file with combination of unzipped files

 val csvRDD = spark.read.options(options).csv("/users/mithun/tsvfilelocation/")

Upvotes: 3

Jacek Laskowski
Jacek Laskowski

Reputation: 74619

All text-related data sources, including CSVDataSource, use Hadoop File API to deal with files (it was in Spark Core's RDDs too).

You can find the relevant lines in readFile that leads to HadoopFileLinesReader which has the following lines:

val fileSplit = new FileSplit(
  new Path(new URI(file.filePath)),
  file.start,
  file.length,
  // TODO: Implement Locality
  Array.empty)

That uses Hadoop's org.apache.hadoop.fs.Path that deals with compression of the underlying file(s).

After quick googling, I was able to find the Hadoop property that deals with compression which is mapreduce.output.fileoutputformat.compress.

That led me to Spark SQL's CompressionCodecs with the following compression configuration:

"none" -> null,
"uncompressed" -> null,
"bzip2" -> classOf[BZip2Codec].getName,
"deflate" -> classOf[DeflateCodec].getName,
"gzip" -> classOf[GzipCodec].getName,
"lz4" -> classOf[Lz4Codec].getName,
"snappy" -> classOf[SnappyCodec].getName)

Below in the code, you can find setCodecConfiguration that uses "our" option.

  def setCodecConfiguration(conf: Configuration, codec: String): Unit = {
    if (codec != null) {
      conf.set("mapreduce.output.fileoutputformat.compress", "true")
      conf.set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString)
      conf.set("mapreduce.output.fileoutputformat.compress.codec", codec)
      conf.set("mapreduce.map.output.compress", "true")
      conf.set("mapreduce.map.output.compress.codec", codec)
    } else {
      // This infers the option `compression` is set to `uncompressed` or `none`.
      conf.set("mapreduce.output.fileoutputformat.compress", "false")
      conf.set("mapreduce.map.output.compress", "false")
    }
  }

The other method getCodecClassName is used to resolve compression option for JSON, CSV, and text formats.

Upvotes: 4

Related Questions