SparkZip.scala

package flat

import java.io.{BufferedReader, InputStreamReader}
import java.util.zip.{ZipEntry, ZipInputStream}

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.input.PortableDataStream
import org.apache.spark.rdd.RDD

object Main {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("main"))
    val strings = z(sc, "/tmp/zips")
    strings.foreach(println)
  }

  def z(sc: SparkContext, path: String): RDD[String] = {
    sc.binaryFiles(path).flatMap { case (name: String, content: PortableDataStream) =>
      val zis: ZipInputStream = new ZipInputStream(content.open)
      val stream: Stream[ZipEntry] = Stream.continually(zis.getNextEntry).takeWhile(_ != null)
      val brs: Stream[BufferedReader] = stream.map(_ => new BufferedReader(new InputStreamReader(zis)))
      val lines: Stream[String] = brs.flatMap(br => Stream.continually(br.readLine()).takeWhile(_ != null).toList)
      lines
    }
  }
}