【问题标题】：Spark/Scala Opening Zipped CSV FilesSpark/Scala 打开压缩的 CSV 文件
【发布时间】：2014-03-18 22:58:22
【问题描述】：

我是 Spark 和 Scala 的新手。我们将广告事件日志文件格式化为 CSV，然后使用 pkzip 进行压缩。我已经看到了很多关于如何使用 Java 解压缩压缩文件的示例，但是我将如何使用 Scala for Spark 来做到这一点？最终，我们希望从每个传入文件中获取、提取数据并将其加载到 Hbase 目标表中。也许这可以用 HadoopRDD 来完成？在此之后，我们将引入 Spark 流来监视这些文件。

谢谢，本

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

在 Spark 中，如果您的文件具有正确的文件名后缀（例如 .gz 表示 gzipped），并且受 org.apache.hadoop.io.compress.CompressionCodecFactory 支持，那么您可以使用

sc.textFile(path)

更新：在撰写本文时，他们是 Hadoop bzip2 库中的一个错误，这意味着尝试使用 spark 读取 bzip2 文件会导致奇怪的异常 - 通常是 ArrayIndexOutOfBounds。

【讨论】：

这对我不起作用。我有一个压缩文件（带有 .zip 扩展名）并且执行 sc.textFile(path) 会引发异常...

【解决方案2】：

默认压缩支持

@samthebest 答案是正确的，如果您使用的是 Spark (Hadoop) 中默认可用的压缩格式。分别是：

bzip2
压缩包
lz4
活泼

我已经在我的另一个答案中更深入地解释了这个主题：https://stackoverflow.com/a/45958182/1549135

阅读压缩包

但是，如果您尝试读取 zip 文件，则需要创建自定义解决方案。我已经提供的答案中提到了一个。

如果您需要从存档中读取多个文件，您可能会对我提供的答案感兴趣：https://stackoverflow.com/a/45958458/1549135

基本上，一直使用sc.binaryFiles，然后解压缩PortableDataStream，就像在示例中一样：

sc.binaryFiles(path, minPartitions)
  .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)
          }

【讨论】：