Amazon web services – Unzipping multiple files from 1 zip files using EMR

The following is a solution in scala. I had to do this before in my job. So I am extracting the relevant bits here.

Few important points to keep in mind.

If possible in your workflow, try to do a tar.gz of your files instead of zip. Because I tried it only with that format.

Secondly, repartition the rdd numPartitionsProvided to a suitably large number so that all your executors are utilized.

ZipFileReader.scala

import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream

import org.apache.spark.input.PortableDataStream
import org.apache.spark.sql.SparkSession

import java.nio.charset.StandardCharsets
import scala.util.Try

object ZipFileReader {

  def decode(bytes: Array[Byte]) =
    new String(bytes, StandardCharsets.UTF_8)


  def extractFilesCSV(ps: PortableDataStream, n: Int = 1024) = Try {
    val tar = new TarArchiveInputStream(new GzipCompressorInputStream(ps.open))
    Stream.continually(Option(tar.getNextTarEntry))
      // Read until next exntry is null
      .takeWhile(_.isDefined)
      // flatten
      .flatMap(x => x)
      // Drop directories
      .filter(!_.isDirectory)
      .filter(ele => ele.getName.split("/")(1).endsWith(".csv"))
      .map(e => {
        println("This could be name", e.getName)
        (e.getName.split("/")(1),
          Stream.continually {
              // Read n bytes
              val buffer = Array.fill[Byte](n)(-1)
              val i = tar.read(buffer, 0, n)
              (i, buffer.take(i))}
            // Take as long as we have read something
            .takeWhile(_._1 > 0)
            .map(_._2)
            .flatten
            .toArray)})
      .toArray
  }


  def extractFilesJSON(ps: PortableDataStream, n: Int = 1024) = Try {
    val tar = new TarArchiveInputStream(new GzipCompressorInputStream(ps.open))
    Stream.continually(Option(tar.getNextTarEntry))
      // Read until next exntry is null
      .takeWhile(_.isDefined)
      // flatten
      .flatMap(x => x)
      // Drop directories
      .filter(!_.isDirectory)
      .filter(ele => ele.getName.split("/")(1).endsWith(".json"))
      .map(e => {
        println("This could be name", e.getName)
        (e.getName.split("/")(1),
          Stream.continually {
              // Read n bytes
              val buffer = Array.fill[Byte](n)(-1)
              val i = tar.read(buffer, 0, n)
              (i, buffer.take(i))}
            // Take as long as we have read something
            .takeWhile(_._1 > 0)
            .map(_._2)
            .flatten
            .toArray)})
      .toArray
  }


  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    val datafilename = "/zip-loc/myfirst.tar.gz"
    val output_path = "/zip-loc/output1/"




    val readInRawCSVFiles = spark.sparkContext.binaryFiles(datafilename).flatMapValues(x =>
      extractFilesCSV(x).toOption).map(ele => ele._2.map(inner_ele => (inner_ele._1, decode(inner_ele._2) )))

    val rawDataCSVParallelized = spark.sparkContext.parallelize(readInRawCSVFiles.first())

    //make this a large number so that all your executors are utilized
    val numPartitionsProvided = 3
    println("Writing out the files to output_path")
    rawDataCSVParallelized.repartition(numPartitions = numPartitionsProvided).map(ele => ele._2).saveAsTextFile(output_path)


    // The following code is only for sanity checking and testing everything is going smoothly
    // The following code can be removed from production file.
    rawDataCSVParallelized.foreach(ele => println("fileName within .tar.gz", ele._1))

    val countRawDataCSVParallelized = rawDataCSVParallelized.count()
    println(s"Count of the csv input files = $countRawDataCSVParallelized")


    val dataCollected = rawDataCSVParallelized.collect()
    val username1_csv = dataCollected.filter(ele => ele._1.equalsIgnoreCase("username1.csv")).head
    println("username1_csv")
    println(username1_csv)
    val username2_csv = dataCollected.filter(ele => ele._1.equalsIgnoreCase("username2.csv")).head
    println("username2_csv")
    println(username2_csv)

  }

}

I have created a tar.gz file using the following command.

tar -czvf myfirst.tar.gz username_files

username_files folder contains the following files as example.

username1.csv
username2.csv
username3.csv
username4.csv
username5.csv
username6.csv

Upload the myfirst.tar.gz to hdfs locally to try it out and check if everything is working ok.