在 Python Spark 中查看 RDD 内容？

Posted 2023-04-15

【中文标题】在 Python Spark 中查看 RDD 内容？

【英文标题】：View RDD contents in Python Spark?

【发布时间】：2014-08-13 20:13:50

【问题描述】：

在 pyspark 中运行一个简单的应用程序。

f = sc.textFile("README.md")
wc = f.flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1)).reduceByKey(add)

我想使用 foreach 操作查看 RDD 内容：

wc.foreach(print)

这会引发语法错误：

SyntaxError: invalid syntax

我错过了什么？

【问题讨论】：

如果你没有设置使用foreach()，你可以使用：for row in f.take(f.count()): print(row)

【参考方案1】：

此错误是因为print 不是 Python 2.6 中的函数。

您可以定义执行打印的辅助 UDF，或使用 __future__ 库将 print 视为函数：

>>> from operator import add
>>> f = sc.textFile("README.md")
>>> def g(x):
...     print x
...
>>> wc.foreach(g)

或

>>> from __future__ import print_function
>>> wc.foreach(print)

但是，我认为使用collect() 将RDD 内容带回驱动程序会更好，因为foreach 在工作节点上执行并且输出可能不一定出现在您的驱动程序/外壳中（它可能将在local 模式下，但在集群上运行时不会）。

>>> for x in wc.collect():
...     print x

【讨论】：

太好了，正如你所说，我使用了 collect()。谢谢，乔希。

收集是要走的路！

【参考方案2】：

在 Spark 2.0 中（我没有使用早期版本进行测试）。简单地说：

print myRDD.take(n)

在您的情况下，n 是行数，myRDD 是 wc。

【参考方案3】：

试试这个：

data = f.flatMap(lambda x: x.split(' '))
map = data.map(lambda x: (x, 1))
mapreduce = map.reduceByKey(lambda x,y: x+y)
result = mapreduce.collect()

请注意，当您运行 collect() 时，RDD（分布式数据集）会在驱动程序节点处聚合，并且本质上会转换为列表。所以很明显，collect() 一个 2T 数据集并不是一个好主意。如果您只需要 RDD 中的几个样本，请使用 take(10)。

【讨论】：

请将此评论移至您的答案，因为没有此评论，您的上述答案是不完整的

【参考方案4】：

如果您想查看 RDD 的内容，那么 yes collect 是一种选择，但它会将所有数据获取到驱动程序，因此可能会出现问题

<rdd.name>.take(<num of elements you want to fetch>)

如果你只想看一个样本就更好了

运行 foreach 并尝试打印，我不建议这样做，因为如果您在集群上运行它，那么打印日志将是执行器本地的，并且它将打印该执行器可访问的数据. print 语句没有改变状态，因此它在逻辑上没有错误。要获取所有日志，您必须执行类似的操作

**Pseudocode**
collect
foreach print

但这可能会导致作业失败，因为收集驱动程序上的所有数据可能会使它崩溃。我建议使用 take 命令，或者如果你想分析它然后使用 sample 在驱动程序上收集或写入文件然后分析它。

【参考方案5】：

您可以简单地收集整个 RDD（这将返回一个行列表）并打印该列表：

print(wc.collect())

【参考方案6】：

通过最新的文档，你可以在驱动上使用rdd.collect().foreach(println)来显示所有，但是可能会导致驱动内存问题，最好是使用rdd.take(desired_number)

https://spark.apache.org/docs/2.2.0/rdd-programming-guide.html

要打印驱动程序上的所有元素，可以使用 collect() 方法首先将 RDD 带到驱动程序节点：rdd.collect().foreach(println)。但是，这可能会导致驱动程序耗尽内存，因为 collect() 将整个 RDD 获取到单个机器；如果只需要打印 RDD 的几个元素，更安全的方法是使用 take()：rdd.take(100).foreach(println)。