在Python的Spark中查看RDD内容？

Question

在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

我错过了什么？

- lmart999

---

如果您不一定要使用 foreach()，您可以使用以下代码：for row in f.take(f.count()): print(row)。 - emmagras

Answer 1

在Spark 2.0中（我没有测试过早期版本），简单地说：

print myRDD.take(n)

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

- Frederico Oliveira

Answer 2

这个错误是因为在Python 2.6中print不是一个函数。

你可以定义一个辅助的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在工作节点上执行，输出可能不会出现在您的驱动程序/ shell中（在local模式下可能会出现，但在集群上运行时不会）。

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

- Josh Rosen

---

太好了，正如你建议的那样，我使用了collect()。谢谢，Josh。 - lmart999

收集是正确的方式！ - Michael David Watson

Answer 3

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

print(wc.collect())

- alehresmann

Answer 4

试试这个：

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（分布式数据集）将在驱动节点聚合，并且基本上转换为列表。因此，显然不建议收集2T数据集。如果您只需要从RDD中获取一些样本，请使用take(10)。

- Jeevs

---

请将这个注释移到你的答案中，因为你上面的答案如果没有这个注释是不完整的。 - iec2011007

Answer 5

如果您想查看RDD的内容，那么collect是一个选项，但它会将所有数据提取到驱动程序中，可能会出现问题。

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

如果您只想查看示例，则最好。

运行 foreach 并尝试打印，我不建议这样做，因为如果您在集群上运行此操作，则打印日志将仅对执行器本地，并且它将打印对该执行器可访问的数据。 print 语句不会改变状态，因此逻辑上是正确的。要获取所有日志，您需要执行类似以下操作：

**Pseudocode**
collect
foreach print

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

- iec2011007

Answer 6

根据最新的文档，您可以在驱动程序上使用rdd.collect().foreach(println)来显示所有内容，但这可能会导致驱动程序内存问题。最好使用rdd.take(desired_number)来显示所需数量的元素。

要在驱动程序上打印所有元素，可以使用collect()方法将RDD首先传输到驱动程序节点，如下：rdd.collect().foreach(println)。然而，这可能会导致驱动程序内存不足，因为collect()将整个RDD获取到单个机器上；如果您只需要打印RDD的少数元素，则更安全的方法是使用take()：rdd.take(100).foreach(println)。

参考链接：https://spark.apache.org/docs/2.2.0/rdd-programming-guide.html

- YDD9