Reading in multiple files compressed in tar.gz archive into Spark [duplicate]

by

A solution is given in Read whole text files from a compression in Spark .
Using the code sample provided, I was able to create a DataFrame from the compressed archive like so:

val jsonRDD = sc.binaryFiles("gzarchive/*").
               flatMapValues(x => extractFiles(x).toOption).
               mapValues(_.map(decode())

val df = sqlContext.read.json(jsonRDD.map(_._2).flatMap(x => x))

This method works fine for tar archives of a relatively small size, but is not suitable for large archive sizes.

A better solution to the problem seems to be to convert the tar archives to Hadoop SequenceFiles, which are splittable and hence can be read and processed in parallel in Spark (as opposed to tar archives.)

See: A Million Little Files – Digital Digressions by Stuart Sierra.

More Related Contents:

  1. How does HashPartitioner work?
  2. How to convert rdd object to dataframe in spark
  3. (Why) do we need to call cache or persist on a RDD
  4. Case class equality in Apache Spark
  5. Spark performance for Scala vs Python
  6. Stackoverflow due to long RDD Lineage
  7. How to find spark RDD/Dataframe size?
  8. Modify collection inside a Spark RDD foreach
  9. Parsing multiline records in Scala
  10. Explanation of fold method of spark RDD
  11. Why does partition parameter of SparkContext.textFile not take effect?
  12. Why does Spark RDD partition has 2GB limit for HDFS?
  13. How to sort an RDD in Scala Spark?
  14. How to transpose an RDD in Spark
  15. Apache Spark – Scala – ReduceByKey – with keys repeating up to twice only
  16. Spark – load CSV file as DataFrame?
  17. How to define partitioning of DataFrame?
  18. Dropping a nested column from Spark DataFrame
  19. Spark dataframe write method writing many small files
  20. Filling gaps in timeseries Spark
  21. java.lang.NoClassDefFoundError: org/apache/spark/streaming/twitter/TwitterUtils$ while running TwitterPopularTags
  22. How to use COGROUP for large datasets
  23. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  24. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  25. Create new Dataframe with empty/null field values
  26. Spark: Add column to dataframe conditionally
  27. Difference between SparkContext, JavaSparkContext, SQLContext, and SparkSession?
  28. Difference between two rows in Spark dataframe
  29. Run a scala code jar appear NoSuchMethodError:scala.Predef$.refArrayOps
  30. Spark UDF with varargs

Leave a Comment