[大数据]Spark读取zip文件原理与详解

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

import org.apache.spark.input.PortableDataStream

val dataAndPortableRDD = sc.binaryFiles("zipData path")

val dataRDD = dataAndPortableRDD.flatMap { case (name: String, content: PortableDataStream) =>
  val zis = new ZipInputStream(content.open)
  Stream.continually(zis.getNextEntry)
    .takeWhile(_ != null)
    .flatMap { _ =>
      //注意，这里非并行
      val br = new BufferedReader(new InputStreamReader(zis))
      Stream.continually(br.readLine()).takeWhile(_ != null)
     //实际上，还要关br和new InputStreamReader(zis)
    }#::: { zis.close; Stream.empty[String] }//不加#::: { zis.close; Stream.empty[String] }会不关闭连接
}

dataRDD.take(10).foreach(println)


//另一种
   sc.binaryFiles(path).repartition(2000)
      .flatMap{
        case (zipFilePath: String, context:PortableDataStream) => {
            val zis = new ZipInputStream(context.open())
            Stream.continually(zis.getNextEntry)
              .takeWhile(_ != null)
              .flatMap(zipEn => {
                if(!zipEn.isDirectory) {
                  scala.io.Source.fromInputStream(zis,"UTF-8").getLines()
                }else {
                  None
                }
              })#::: { zis.close; Stream.empty[String]}
        }
      }.repartition(2000)

问题：

当文件比较大时，效率特别慢，比如：

13G的文件读取，50核300G内存情况下，5h还不能处理完（实际上，是因为没有关闭连接而卡住导致的）

spark2.3的官方文档中也写到：

Partitioning is determined by data locality. This may result in too few partitions by default.

  /** Returns the longest prefix of this `Stream` whose elements satisfy the
   * predicate `p`.
   *
   * @param p the test predicate.
   * @return A new `Stream` representing the values that satisfy the predicate
   * `p`.
   *
   * @example {{{
   + naturalsFrom(0) takeWhile { _ < 5 } mkString ", "
   * produces: "0, 1, 2, 3, 4"
   * }}}
   */
  override def takeWhile(p: A => Boolean): Stream[A] =
    if (!isEmpty && p(head)) cons(head, tail takeWhile p)
    else Stream.Empty

binaryFile：https://spark.apache.org/docs/2.0.2/api/scala/index.html#org.apache.spark.SparkContext

zipInputStream:

https://docs.oracle.com/javase/7/docs/api/java/util/zip/ZipInputStream.html\

PortableDataStream:

https://spark.apache.org/docs/2.3.0/api/scala/index.html#org.apache.spark.input.PortableDataStream

Stream:

https://zhuanlan.zhihu.com/p/130958554

分区问题：

https://www.coder.work/article/6527071

不关闭会出问题，表现为大文件时处理完后卡住不动

https://stackoverflow.com/questions/32080475/how-to-read-a-zip-containing-multiple-files-in-apache-spark

Spark 2.4+?，问题应该得到解决，请参阅此答案下方的@Rahul 评论。

Spark 2.1-2.3?，minPartitions?的?binaryFiles()?参数被忽略。请参阅?Spark-16575?和?commit changes to function setMinPartitions()?。请注意提交更改如何在函数中不再使用?minPartitions!

如果您使用?binaryFiles()?读取多个二进制文件，则输入文件将根据以下内容合并到分区中:

• spark.files.maxPartitionBytes?，默认 128 MB
• spark.files.openCostInBytes?，默认 4 MB
• spark.default.parallelism
• 输入的总大小
• here?描述了前三个配置项。查看上面的提交更改以查看实际计算。
  
  我有一个场景，我希望每个输入分区最多 40 MB，因此每个任务 40 MB……以在解析时增加并行度。 (Spark 将 128 MB 放入每个分区，降低了我的应用程序的速度。)我在调用?spark.files.maxPartitionBytes?之前将?binaryFiles()?设置为 40 M:

  spark = SparkSession \
     .builder \
     .config("spark.files.maxPartitionBytes", 40*1024*1024)
  

  
  对于只有一个输入文件，@user9864979 的答案是正确的:不能仅使用?binaryFiles()?将单个文件拆分为多个分区。
  
  使用?Spark 1.6?读取多个文件时，minPartitions?参数确实有效，您必须使用它。如果不这样做，您将遇到?Spark-16575?问题:您的所有输入文件将仅被读入两个分区!
  
  您会发现 Spark 通常会为您提供比您要求的更少的输入分区。我有一个场景，我希望每两个输入二进制文件有一个输入分区。我发现将?minPartitions?设置为“输入文件数 * 7/10”大致可以满足我的需求。我有另一种情况，我希望每个输入文件有一个输入分区。我发现将?minPartitions?设置为“输入文件数 * 2”给了我想要的。
  
  Spark 1.5?binaryFiles()?的行为:每个输入文件都有一个分区。