【问题标题】：How to read a zip containing multiple files in Apache Spark如何在 Apache Spark 中读取包含多个文件的 zip
【发布时间】：2015-11-11 20:47:10
【问题描述】：

我有一个包含多个文本文件的压缩文件。我想读取每个文件并构建一个包含每个文件内容的 RDD 列表。

val test = sc.textFile("/Volumes/work/data/kaggle/dato/test/5.zip")

将只是整个文件，但是如何遍历 zip 的每个内容，然后使用 Spark 将其保存在 RDD 中。

我对 Scala 或 Python 很好。

在 Python 中使用 Spark 的可能解决方案 -

archive = zipfile.ZipFile(archive_path, 'r')
file_paths = zipfile.ZipFile.namelist(archive)
for file_path in file_paths:
    urls = file_path.split("/")
    urlId = urls[-1].split('_')[0]

【问题讨论】：

嗨@AbhishekChoudhary - 以下哪种解决方案最适合您？谢谢。
使用spark API，读取单个RDD中保存的所有文件，然后使用不同的过滤机制对数据进行分区
解压缩文件本质上是一个单线程过程——在 Spark 中这样做不是浪费资源吗？
曾经是，但现在 API 也可以在 Spark 中读取压缩文件。

标签： scala apache-spark pyspark

【解决方案1】：

Apache Spark 默认压缩支持

我已经在其他答案中写了所有必要的理论，您可能想参考：https://stackoverflow.com/a/45958182/1549135

读取包含多个文件的 zip

我已经听从了 @Herman 的建议并使用了ZipInputStream。这给了我这个解决方案，它返回 zip 内容的 RDD[String]。

import java.io.{BufferedReader, InputStreamReader}
import java.util.zip.ZipInputStream
import org.apache.spark.SparkContext
import org.apache.spark.input.PortableDataStream
import org.apache.spark.rdd.RDD

implicit class ZipSparkContext(val sc: SparkContext) extends AnyVal {

    def readFile(path: String,
                 minPartitions: Int = sc.defaultMinPartitions): RDD[String] = {

      if (path.endsWith(".zip")) {
        sc.binaryFiles(path, minPartitions)
          .flatMap { case (name: String, content: PortableDataStream) =>
            val zis = new ZipInputStream(content.open)
            Stream.continually(zis.getNextEntry)
                  .takeWhile {
                      case null => zis.close(); false
                      case _ => true
                  }
                  .flatMap { _ =>
                      val br = new BufferedReader(new InputStreamReader(zis))
                      Stream.continually(br.readLine()).takeWhile(_ != null)
                  }
        }
      } else {
        sc.textFile(path, minPartitions)
      }
    }
  }

只需导入隐式类并调用 SparkContext 上的 readFile 方法即可：

import com.github.atais.spark.Implicits.ZipSparkContext
sc.readFile(path)

【讨论】：

你没有关闭连接。
@Programmer 我尝试关闭它，但是这种方法对我来说失败了。所以我把它留给了 Spark。
@Atais Spark 在我的情况下没有关闭流。我试图从 S3 读取数千个文件，但由于连接线程池耗尽而失败，但一旦我关闭代码中的流，它就可以工作。无论如何，及时进行清理总是一个好主意。
您能创建一个答案吗？您是如何处理的？还是贴在某个地方？
我还有其他问题。完成后会发布。是这样的stackoverflow.com/questions/35746539/close-a-stream

【解决方案2】：

如果您正在阅读二进制文件，请使用sc.binaryFiles。这将返回一个包含文件名和PortableDataStream 的元组RDD。您可以将后者输入ZipInputStream。

【讨论】：

【解决方案3】：

这是@Atais 解决方案的工作版本（需要通过关闭流来增强）：

implicit class ZipSparkContext(val sc: SparkContext) extends AnyVal {

def readFile(path: String,
             minPartitions: Int = sc.defaultMinPartitions): RDD[String] = {

  if (path.toLowerCase.contains("zip")) {

    sc.binaryFiles(path, minPartitions)
      .flatMap {
        case (zipFilePath, zipContent) ⇒
          val zipInputStream = new ZipInputStream(zipContent.open())
          Stream.continually(zipInputStream.getNextEntry)
            .takeWhile(_ != null)
            .map { _ ⇒
              scala.io.Source.fromInputStream(zipInputStream, "UTF-8").getLines.mkString("\n")
            } #::: { zipInputStream.close; Stream.empty[String] }
      }
  } else {
    sc.textFile(path, minPartitions)
  }
}
}

那么您只需执行以下操作即可读取 zip 文件：

sc.readFile(path)

【讨论】：

如何将文件名添加到输出以便我可以过滤文件名想象一个 zip 文件有多个模式文件如果我可以在 rdd 中获取文件名，我可以在文件名上使用 spark input_file_name 虚拟列@mahmoud mehdi
这也会给出文件名，.map { x ⇒ val filename1 = x.getName scala.io.Source.fromInputStream(zipInputStream, "UTF-8").getLines.mkString(s"~ ${filename1}\n")+s"~${filename1}" } #::: { zipInputStream.close; Stream.empty[String] }

【解决方案4】：

这只会过滤第一行。任何人都可以分享您的见解。我正在尝试读取压缩的 CSV 文件并创建 JavaRDD 以进行进一步处理。

JavaPairRDD<String, PortableDataStream> zipData =
                sc.binaryFiles("hdfs://temp.zip");
        JavaRDD<Record> newRDDRecord = zipData.flatMap(
          new FlatMapFunction<Tuple2<String, PortableDataStream>, Record>(){
              public Iterator<Record> call(Tuple2<String,PortableDataStream> content) throws Exception {
                  List<Record> records = new ArrayList<Record>();
                      ZipInputStream zin = new ZipInputStream(content._2.open());
                      ZipEntry zipEntry;
                      while ((zipEntry = zin.getNextEntry()) != null) {
                          count++;
                          if (!zipEntry.isDirectory()) {
                              Record sd;
                              String line;
                              InputStreamReader streamReader = new InputStreamReader(zin);
                              BufferedReader bufferedReader = new BufferedReader(streamReader);
                              line = bufferedReader.readLine();
                              String[] records= new CSVParser().parseLineMulti(line);
                              sd = new Record(TimeBuilder.convertStringToTimestamp(records[0]),
                                        getDefaultValue(records[1]),
                                        getDefaultValue(records[22]));
                              records.add(sd);
                          }
                      }

                return records.iterator();
              }

        });

【讨论】：

【解决方案5】：

这是另一种可行的解决方案，它给出了文件名，以后可以拆分并用于从中创建单独的架构。

implicit class ZipSparkContext(val sc: SparkContext) extends AnyVal {

    def readFile(path: String,
                 minPartitions: Int = sc.defaultMinPartitions): RDD[String] = {

      if (path.toLowerCase.contains("zip")) {

        sc.binaryFiles(path, minPartitions)
          .flatMap {
            case (zipFilePath, zipContent) ⇒
              val zipInputStream = new ZipInputStream(zipContent.open())
              Stream.continually(zipInputStream.getNextEntry)
                .takeWhile(_ != null)
                .map { x ⇒
                  val filename1 = x.getName
                  scala.io.Source.fromInputStream(zipInputStream, "UTF-8").getLines.mkString(s"~${filename1}\n")+s"~${filename1}"
                } #::: { zipInputStream.close; Stream.empty[String] }
          }
      } else {
        sc.textFile(path, minPartitions)
      }
    }
  }

完整代码在这里

https://github.com/kali786516/Spark2StructuredStreaming/blob/master/src/main/scala/com/dataframe/extraDFExamples/SparkReadZipFiles.scala

【讨论】：