如何打印RDD的内容？答案

【问题标题】：How to print the contents of RDD?如何打印RDD的内容？
【发布时间】：2014-06-04 02:33:17
【问题描述】：

我正在尝试将集合的内容打印到 Spark 控制台。

我有一个类型：

linesWithSessionId: org.apache.spark.rdd.RDD[String] = FilteredRDD[3]

我使用命令：

scala> linesWithSessionId.map(line => println(line))

但这是打印出来的：

res1: org.apache.spark.rdd.RDD[Unit] = MappedRDD[4] at map at :19

如何将 RDD 写入控制台或将其保存到磁盘以便查看其内容？

【问题讨论】：

嗨！您是否阅读了答案中的 cmets 并被您接受？这似乎具有误导性
@dk14 同意，我已重新分配接受的答案
RDD 被降级为二等公民，您应该使用 DataFrame 和 show 方法。

标签： scala apache-spark

【解决方案1】：

在java语法中：

rdd.collect().forEach(line -> System.out.println(line));

【讨论】：

【解决方案2】：

map 函数是一个transformation，这意味着在您对其运行action 之前，Spark 不会真正评估您的 RDD。

要打印它，你可以使用foreach（这是一个动作）：

linesWithSessionId.foreach(println)

要将其写入磁盘，您可以使用RDD API 中的saveAs... 函数之一（静态操作）

【讨论】：

也许你需要提到collect，以便在控制台中打印RDD。
foreach 本身将首先“物化”RDD，然后在每个元素上运行println，所以这里实际上并不需要collect（当然你可以使用它）。 .
其实没有collect()，在foreach之前，控制台什么都看不到。
实际上它在我的 Spark shell 中运行良好，即使在 1.2.0 中也是如此。但我想我知道这种混淆来自哪里：最初的问题是询问如何将 RDD 打印到 Spark 控制台（= shell），所以我假设他会运行本地作业，在这种情况下 foreach 工作正常。如果您在集群上运行作业并且想要打印您的 rdd，那么您应该 collect（正如其他 cmets 和答案所指出的那样）以便在执行 println 之前将其发送给驱动程序。如果您的 RDD 太大，使用 Oussama 建议的 take 可能是一个好主意。
以上答案不好。你应该不接受它。 Foreach 不会打印到控制台，它会在您的工作节点上打印。如果您只有一个节点，则 foreach 将起作用。但是如果你只有一个节点，那你为什么要使用 spark 呢？只需使用 SQL awk、Grep 或其他更简单的东西。所以我认为唯一有效的答案是收集。如果 collect 对您来说太大了，并且您只想要一个示例，使用 take 或 head 或类似功能，如下所述。

【解决方案3】：

您可以不必每次都输入；

[1] 在 Spark Shell 中创建一个通用打印方法。

def p(rdd: org.apache.spark.rdd.RDD[_]) = rdd.foreach(println)

[2] 甚至更好的是，使用隐式，您可以将函数添加到 RDD 类中以打印其内容。

implicit class Printer(rdd: org.apache.spark.rdd.RDD[_]) {
    def print = rdd.foreach(println)
}

示例用法：

val rdd = sc.parallelize(List(1,2,3,4)).map(_*2)

p(rdd) // 1
rdd.print // 2

输出：

重要

仅当您在本地模式下工作并使用少量数据集时才有意义。否则，您将无法在客户端看到结果，或者由于大数据集结果而耗尽内存。

【讨论】：

【解决方案4】：

c.take(10)

Spark 较新版本会很好地显示表格。

【讨论】：

【解决方案5】：

您可以将您的RDD 转换为DataFrame 然后show() 它。

// For implicit conversion from RDD to DataFrame
import spark.implicits._

fruits = sc.parallelize([("apple", 1), ("banana", 2), ("orange", 17)])

// convert to DF then show it
fruits.toDF().show()

这将显示数据的前 20 行，因此数据的大小应该不是问题。

+------+---+                                                                    
|    _1| _2|
+------+---+
| apple|  1|
|banana|  2|
|orange| 17|
+------+---+

【讨论】：

我认为是import spark.implicits._
这里使用的库是什么？我无法在 spark 范围内检测到 toDF 和 spark.implicits._。

【解决方案6】：

您也可以另存为文件：rdd.saveAsTextFile("alicia.txt")

【讨论】：

【解决方案7】：

在python中

   linesWithSessionIdCollect = linesWithSessionId.collect()
   linesWithSessionIdCollect

这将打印出RDD的所有内容

【讨论】：

谢谢，但我用 scala 而不是 python 标记了这个问题

【解决方案8】：

myRDD.foreach(println) 和myRDD.collect().foreach(println) 之间可能存在许多架构差异（不仅是“收集”，还有其他操作）。我看到的差异之一是在执行myRDD.foreach(println) 时，输出将是随机顺序。例如：如果我的 rdd 来自每行都有一个数字的文本文件，则输出将具有不同的顺序。但是当我执行myRDD.collect().foreach(println) 时，顺序仍然和文本文件一样。

【讨论】：

【解决方案9】：

如果你想查看某个RDD的内容，一种方法是使用collect()：

myRDD.collect().foreach(println)

但是，当 RDD 有数十亿行时，这不是一个好主意。使用take() 打印几张：

myRDD.take(n).foreach(println)

【讨论】：

如果我在 RDD（有数百万行）上使用 foreach 将内容作为单个文件写入 HDFS，它会在集群上正常工作吗？
我不在RDD上使用saveAsTextFile的原因是，我需要将RDD内容写入多个文件，这就是我使用foreach的原因
如果你想保存在一个文件中，你可以在调用 saveAsTextFile 之前将你的 RDD 合并到一个分区中，但这同样可能会导致问题。我认为最好的选择是在 HDFS 中写入多个文件，然后使用 hdfs dfs --getmerge 合并文件
您说当在 RDD 上使用 foreach 时，它会将其持久化到驱动程序的 RAM 中，该语句是否正确？因为我的理解是 foreach 将在每个工人[集群]上运行而不是在驱动程序上运行。
saveAsTextFile 将为每个分区写入一个文件，这就是您想要的（多个文件）。否则，正如 Oussama 建议的那样，您可以执行 rdd.coalesce(1).saveAsTextFile() 来获取一个文件。如果 RDD 的分区太少，你可以试试 rdd.repartition(N).saveAsTextFile()

【解决方案10】：

如果您在集群上运行它，那么 println 将不会打印回您的上下文。您需要将RDD 数据带到您的会话中。为此，您可以将其强制到本地数组，然后将其打印出来：

linesWithSessionId.toArray().foreach(line => println(line))

【讨论】：