如何使saveAsTextFile不将输出拆分为多个文件？

Question

在Spark中使用Scala时，每当我使用saveAsTextFile转储结果时，它似乎将输出分成多个部分。我只是向它传递一个参数（路径）。

val year = sc.textFile("apat63_99.txt").map(_.split(",")(1)).flatMap(_.split(",")).map((_,1)).reduceByKey((_+_)).map(_.swap)
year.saveAsTextFile("year")

1. 输出数量是否与其使用的减速器数量相对应
2. 这是否意味着输出被压缩
3. 我知道我可以使用bash将输出组合在一起，但是有没有一个选项可以将输出存储在单个文本文件中，而不进行拆分？？我看了API文档，但它并没有说太多

Answer 1

您可以调用coalesce（1），然后调用saveAsTextFile（）——但如果您有大量数据，这可能是个坏主意。每次拆分都会生成单独的文件，就像在Hadoop中一样，以便让单独的映射器和还原器写入不同的文件。只有当您的数据很少时，才有一个单独的输出文件，在这种情况下，您还可以像@aaronman所说的那样执行collect（）。

Answer 2

对于使用较大数据集的用户：

>

rdd.collect（）在这种情况下不应该使用，因为它会将所有数据作为驱动程序中的Array收集，这是获取内存溢出的最简单方法。

<代码>rdd。聚结（1）。也不应使用saveAsTextFile（），因为在存储数据的单个节点上执行上游阶段的并行将丢失。

<代码>rdd。合并（1，shuffle=true）。saveAsTextFile（）是最好的简单选项，因为它将使上游任务的处理保持并行，然后只对一个节点执行无序移动（rdd.repartition（1））。saveAsTextFile（）是一个确切的同义词）。

<代码>rdd。下面提供的saveAsSingleTextFile（）还允许使用特定名称将rdd存储在单个文件中，同时保留rdd的并行性属性。合并（1，shuffle=true）。saveAsTextFile（）。

rdd.coalesce（1， shuffle=true）. saveAsTextFile（"path/to/file.txt"）可能不方便的是，它实际上生成的文件的路径是path/to/file.txt/part-00000而不是path/to/file.txt。

以下解决方案是rdd。saveAsSingleTextFile（“path/to/file.txt”）将实际生成一个路径为path/to/file的文件。txt文件：

package com.whatever.package

import org.apache.spark.rdd.RDD
import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
import org.apache.hadoop.io.compress.CompressionCodec

object SparkHelper {

  // This is an implicit class so that saveAsSingleTextFile can be attached to
  // SparkContext and be called like this: sc.saveAsSingleTextFile
  implicit class RDDExtensions(val rdd: RDD[String]) extends AnyVal {

    def saveAsSingleTextFile(path: String): Unit =
      saveAsSingleTextFileInternal(path, None)

    def saveAsSingleTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit =
      saveAsSingleTextFileInternal(path, Some(codec))

    private def saveAsSingleTextFileInternal(
        path: String, codec: Option[Class[_ <: CompressionCodec]]
    ): Unit = {

      // The interface with hdfs:
      val hdfs = FileSystem.get(rdd.sparkContext.hadoopConfiguration)

      // Classic saveAsTextFile in a temporary folder:
      hdfs.delete(new Path(s"$path.tmp"), true) // to make sure it's not there already
      codec match {
        case Some(codec) => rdd.saveAsTextFile(s"$path.tmp", codec)
        case None        => rdd.saveAsTextFile(s"$path.tmp")
      }

      // Merge the folder of resulting part-xxxxx into one file:
      hdfs.delete(new Path(path), true) // to make sure it's not there already
      FileUtil.copyMerge(
        hdfs, new Path(s"$path.tmp"),
        hdfs, new Path(path),
        true, rdd.sparkContext.hadoopConfiguration, null
      )
      // Working with Hadoop 3?: https://stackoverflow.com/a/50545815/9297144

      hdfs.delete(new Path(s"$path.tmp"), true)
    }
  }
}

可以这样使用：

import com.whatever.package.SparkHelper.RDDExtensions

rdd.saveAsSingleTextFile("path/to/file.txt")

scala prettyprint-override">// Or if the produced file is to be compressed:
import org.apache.hadoop.io.compress.GzipCodec
rdd.saveAsSingleTextFile("path/to/file.txt.gz", classOf[GzipCodec])

此片段：

>

首先将带有rdd.saveAsTextFile（"path/to/file.txt"）的rdd存储在临时文件夹path/to/file.txt.tmp中，就好像我们不想将数据存储在一个文件中一样（这使上游任务的处理保持并行）

然后，仅使用hadoop文件系统api，我们继续对不同的输出文件进行合并（FileUtil.copyMerge（）），以创建最终的输出单个文件路径/到/文件。txt。

Answer 3

它将其保存为多个文件的原因是计算是分布式的。如果输出足够小，以至于您认为可以在一台机器上安装它，那么您可以使用

val arr = year.collect()

然后将生成的数组保存为文件，另一种方法是使用自定义分区器分区通过，并使其所有内容都进入一个分区，但这并不可取，因为您不会得到任何并行化。

如果需要使用saveAsTextFile保存文件，可以使用coalesce（1，true）。saveAsTextFile（）。这基本上意味着进行计算，然后合并到1个分区。您还可以使用重新分区（1），它只是将shuffle参数设置为true时合并的包装。查看RDD的来源。scala是我解决大部分问题的方法，你应该看看。