将存储在tar.gz压缩文件中的多个文件读入Spark

Question

我将尝试从打包的多个json文件中创建一个Spark RDD。例如，我有3个文件。

file1.json
file2.json
file3.json

这些内容包含在archive.tar.gz中。

我想从JSON文件创建一个数据帧。问题是，Spark没有正确地读取JSON文件。使用sqlContext.read.json("archive.tar.gz")或sc.textFile("archive.tar.gz")创建RDD会导致输出混乱/多余。

有没有办法在Spark中处理包含多个文件的gzipped存档？

更新

使用Read whole text files from a compression in Spark答案中给出的方法，我能够让事情运行起来，但是这种方法似乎不适用于大的tar.gz存档（>200 mb压缩），因为应用程序会在大型存档大小上崩溃。由于我处理的一些存档在压缩后达到了2 GB的大小，所以我想知道是否有一种有效的方法来解决这个问题。

我正在尝试避免提取存档然后将文件合并在一起，因为这将耗费时间。

- zenofsahil

---

我遇到了大文件的同样问题，你找到了解决方法吗（除了手动解压缩，然后加载到Spark中）？ :) - ixaxaar

@ixaxaar，我最终将所有的tar归档文件转换为Hadoop序列文件，这样Spark可以很好地处理它们。 https://stuartsierra.com/2008/04/24/a-million-little-files - zenofsahil

非常感谢这个jar文件！实际上我有50个tar文件，每个文件里面都有一百万个小的（json）文件 :D - ixaxaar

那么，这就是了。 :) - zenofsahil

Answer 1

在 Read whole text files from a compression in Spark 中提供了一种解决方案。 使用提供的代码示例，我能够从压缩档案中创建一个 DataFrame ，代码如下：

```python df = spark.read.text("path/to/compressed/archive") ```

val jsonRDD = sc.binaryFiles("gzarchive/*").
               flatMapValues(x => extractFiles(x).toOption).
               mapValues(_.map(decode())

val df = sqlContext.read.json(jsonRDD.map(_._2).flatMap(x => x))

这种方法适用于大小相对较小的tar归档，但不适用于大型归档大小。

解决该问题的更好方案似乎是将tar归档转换为Hadoop SequenceFiles，它们可以分割，因此可以在Spark中并行读取和处理（与tar归档相反）。

参见：A Million Little Files – Digital Digressions by Stuart Sierra。

- zenofsahil

Answer 2

*.tar.gz 文件中的文件，正如您已经提到的那样，都被压缩了。您不能将这三个文件放入一个单独的压缩 tar 文件中，并期望导入函数（只能处理文本数据）知道如何处理解压缩文件、从 tar 存档中解包它们，然后逐个导入每个文件。

建议您花时间手动上传每个单独的 json 文件，因为 sc.textfile 和 sqlcontext.read.json 函数都无法处理压缩数据。

- DJHenjin