读取多个文件并收集是否将它们带到 spark 的驱动程序

Question

代码 sn-p：

val inp = sc.textFile("C:\\mk\\logdir\\foldera\\foldera1\\log.txt").collect.mkString(" ")

我知道上面的代码会读取整个文件并将它们组合成一个字符串并执行它的驱动程序节点（单次执行。不是并行的）。

 val inp = sc.textFile("C:\\mk\\logdir\\*\\*\\log.txt")
 code block{ }
 sc.stop

Q1 ）这里我正在读取多个文件（存在于上述文件夹结构中）。我相信在这种情况下，每个文件都将被创建为分区并被发送到单独的节点并并行执行。我的理解正确吗？有人可以证实这一点吗？或者有什么我可以系统地确认的吗？

val inp = sc.textFile("C:\\mk\\logdir\\*\\*\\log.txt")
val cont = inp.collect.mkString(" ")
 code block{ }
 sc.stop

Q2) spark 如何处理这种情况。虽然我正在收集，但我认为它不会收集所有文件中的所有内容，而只会收集一个文件。我对吗？有人可以帮我理解这一点吗？

非常感谢您的宝贵时间和帮助。

Answer 1

Q1 ）这里我正在读取多个文件（在上面 文件夹结构）。我相信在这种情况下，每个文件都将被创建为 分区 & 将被发送到单独的节点 & 并行执行。我是不是 我的理解正确吗？有人可以证实这一点吗？或者有没有 反正我可以系统地确认吗？

答案：

SparkContext 的 TextFile 方法，即 sc.textFile 创建一个 RDD，每行作为一个元素。如果 data 中有 10 个文件，即yourtextfilesfolder 文件夹，则会创建 10 个分区。您可以通过以下方式验证分区数：

yourtextfilesfolder.partitions.length

但是，分区是由数据局部性决定的。这可能会导致默认分区太少。 AFAIK 不保证会创建一个分区，请参阅“SparkContext.textFile”的代码。

& 'minPartitions' - 建议生成的 RDD 的最小分区数

为了更好地理解，请参阅下面的方法。

/**
           * Read a text file from HDFS, a local file system (available on all nodes), or any
           * Hadoop-supported file system URI, and return it as an RDD of Strings.
           */
          def textFile(
              path: String,
              minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
            assertNotStopped()
            hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
              minPartitions).map(pair => pair._2.toString).setName(path)
          }

you can mention minPartitions as shown above from SparkContext.scala

Q2) spark 如何处理这种情况。虽然我正在收集，但我认为 它不会收集所有文件中的所有内容，而只会收集一个 文件 。我对吗？有人可以帮我理解这一点吗？

回答：您的 rdd 由多个文本文件构成。所以 collect 将从所有分区收集来自不同文件的驱动程序，而不是一次收集一个文件。

您可以验证：使用 rdd.collect

---

但是，如果您想阅读多个文本文件，您也可以使用wholeTextFiles 请参阅下面方法中的@note 小文件是首选，大文件也是允许的，但可能会导致性能不佳。

见spark-core-sc-textfile-vs-sc-wholetextfiles

文档：

RDD> wholeTextFiles(字符串路径，int minPartitions) 从本地文件 HDFS 读取文本文件目录 系统（在所有节点上可用），或任何 Hadoop 支持的文件系统 URI。

/**
   * Read a directory of text files from HDFS, a local file system (available on all nodes), or any
   * Hadoop-supported file system URI. Each file is read as a single record and returned in a
   * key-value pair, where the key is the path of each file, the value is the content of each file.
   *
   * <p> For example, if you have the following files:
   * {{{
   *   hdfs://a-hdfs-path/part-00000
   *   hdfs://a-hdfs-path/part-00001
   *   ...
   *   hdfs://a-hdfs-path/part-nnnnn
   * }}}
   *
   * Do `val rdd = sparkContext.wholeTextFile("hdfs://a-hdfs-path")`,
   *
   * <p> then `rdd` contains
   * {{{
   *   (a-hdfs-path/part-00000, its content)
   *   (a-hdfs-path/part-00001, its content)
   *   ...
   *   (a-hdfs-path/part-nnnnn, its content)
   * }}}
   *
   * @note Small files are preferred, large file is also allowable, but may cause bad performance.
   * @note On some filesystems, `.../path/&#42;` can be a more efficient way to read all files
   *       in a directory rather than `.../path/` or `.../path`
   * @note Partitioning is determined by data locality. This may result in too few partitions
   *       by default.
   *
   * @param path Directory to the input data files, the path can be comma separated paths as the
   *             list of inputs.
   * @param minPartitions A suggestion value of the minimal splitting number for input data.
   * @return RDD representing tuples of file path and the corresponding file content
   */
  def wholeTextFiles(
      path: String,
      minPartitions: Int = defaultMinPartitions): RDD[(String, String)] = withScope {
.....
  }

示例：

val distFile = sc.textFile("data.txt")
Above command returns the content of the file:
scala> distFile.collect()
res16: Array[String] = Array(1,2,3, 4,5,6)


 SparkContext.wholeTextFiles can return (filename, content).
    val distFile = sc.wholeTextFiles("/tmp/tmpdir")

scala> distFile.collect()
res17: Array[(String, String)] =
Array((maprfs:/tmp/tmpdir/data3.txt,"1,2,3
4,5,6
"), (maprfs:/tmp/tmpdir/data.txt,"1,2,3
4,5,6
"), (maprfs:/tmp/tmpdir/data2.txt,"1,2,3
4,5,6
"))

在你的情况下，我更喜欢SparkContext.wholeTextFiles，如果你想要的话，你可以在上面收集后获取文件名及其内容。

Answer 2

Spark 是一种用于大规模数据处理的快速通用引擎。它并行处理所有数据。所以，回答第一个问题，然后是，在以下情况下：

val inp = sc.textFile("C:\\mk\\logdir\\*\\*\\log.txt")
code block{ }
sc.stop

每个文件将被创建为分区并被发送到单独的节点并并行执行。但是，根据文件的大小，分区数可能大于正在处理的文件数。例如，如果folder1 和folder2 中的log.txt 大小为几KB，则只创建2 个分区，因为会有2 个文件，它们将被并行处理。

但是，如果folder1 中的log.txt 具有以GB 为单位的大小，则会为其创建多个分区，并且分区数将大于文件数。

但是，我们始终可以使用repartition() 或coalesce() 方法更改RDD 的分区数。

要回答第二个问题，那么在以下情况下：

val inp = sc.textFile("C:\\mk\\logdir\\*\\*\\log.txt")
val cont = inp.collect.mkString(" ")
code block{ }
sc.stop

Spark 将收集所有文件的内容，而不仅仅是一个文件。因为，collect() 的意思是获取存储的 RDD 中的所有内容，并以集合的形式将其返回给 Driver。