将大文本文件导入 Spark

Question

我有一个 360GB 的管道分隔文本文件，压缩 (gzip)。该文件位于 S3 存储桶中。 这是我第一次使用 Spark。我知道您可以对文件进行分区，以允许多个工作节点对数据进行操作，从而获得巨大的性能提升。但是，我正在尝试找到一种有效的方法将我的一个 360GB 文件转换为分区文件。有没有办法使用多个 Spark 工作节点来处理我的一个压缩文件以对其进行分区？不幸的是，我无法控制我只是得到一个大文件的事实。我可以自己解压缩文件并将其分解为多个文件（例如 360 个 1GB 文件），但我只会使用一台机器来执行此操作，而且速度会很慢。我需要使用 Spark 对数据进行一些昂贵的转换，所以我认为对文件进行分区是必要的。我在 Amazon Glue 中使用 Spark，所以我知道它可以扩展到大量机器。另外，我正在使用 python (pyspark)。

谢谢。

Answer 1

一种可能的解决方案是使用 Amazon's S3DistCp 作为 EMR 集群上的一个步骤，将 360GB 文件复制到集群上可用的 HDFS 文件系统中（这需要在 EMR 上部署 Hadoop）。

S3DistCp 的一个好处是您可以更改输出文件的编解码器，并将原始 gzip 文件转换为允许 Spark 为其 RDD 生成多个分区的格式。

但是我不确定 S3DistCp 执行操作需要多长时间（这是 S3 上的 Hadoop Map/Reduce。从 EMR 运行时，它受益于优化的 S3 库，但我担心 Hadoop 会在生成 Map 任务时面临与 Spark 相同的限制）。

Answer 2

首先我建议你必须使用带有 zlib 压缩的 ORC 格式，这样你就可以得到将近 70% 的压缩率，并且根据我的研究，ORC 是最适合最快数据处理的文件格式。所以你必须加载你的文件并简单地将它写入带有重新分区的orc格式。

df.repartition(500).write.option("compression","zlib").mode("overwrite").save("testoutput.parquet")

Answer 3

如果我没记错的话，如果您使用 SparkContext.textFile 读取文件，Spark 会使用 Hadoop 的 TextInputFormat。如果设置了压缩编解码器，TextInputFormat 通过检查代码是否为SplittableCompressionCodec 的实例来确定文件是否可拆分。

我相信 GZIP 是不可拆分的，Spark 只能生成一个分区来读取整个文件。

你可以做的是：
1. 在SparkContext.textFile 之后添加一个重新分区，这样您至少有一个以上的转换处理部分数据。
2. 请求多个文件，而不仅仅是一个 GZIP 文件
3. 编写一个应用程序，在运行 Spark 应用程序之前将文件解压缩并拆分为多个输出文件。
4. 为 GZIP 编写自己的压缩编解码器（这有点复杂）。

看看这些链接：

TextInputFormat source code for TextInputFormat
GzipCodec source code for GZIPCodec

这些在 java 中，但我确信它们有等效的 Python/Scala 版本。