Pyspark 申请 foreach

Pyspark applying foreach

我是 Pyspark 的菜鸟,我假装玩了几个函数,以便更好地理解如何在更现实的场景中使用它们。有一段时间,我尝试将特定函数应用于 RDD 中的每个数字。我的问题基本上是,当我尝试打印我从 RDD 中获取的内容时,结果是 None

我的代码:

from pyspark import SparkConf , SparkContext

conf = SparkConf().setAppName('test')
sc = SparkContext(conf=conf)

sc.setLogLevel("WARN")


changed = []

def div_two (n):
    opera = n / 2
    return opera

numbers = [8,40,20,30,60,90]

numbersRDD = sc.parallelize(numbers)

changed.append(numbersRDD.foreach(lambda x: div_two(x)))

#result = numbersRDD.map(lambda x: div_two(x))

for i in changed:
    print(i) 

我很高兴能清楚地解释为什么这会出现在列表中,以及使用 foreach 实现该目的的正确方法是否可行。

感谢

你对 div_two 的函数定义看起来不错,但还可以简化为

def div_two (n):
    return n/2

并且您已将 整数数组 转换为 rdd,这也很好。

主要问题是您试图将 rdds 添加到使用 [=15 更改的 array =]函数。但是如果你看看 foreach

的定义
def foreach(self, f) Inferred type: (self: RDD, f: Any) -> None

表示 return 类型是 None。这就是打印的内容。

您不需要数组变量来打印 RDD 的已更改 元素 。您可以简单地编写一个用于打印的函数并在 foreach 函数

中调用该函数
def printing(x):
    print x

numbersRDD.map(div_two).foreach(printing)

您应该打印结果。

您仍然可以将 rdd 添加到 array 变量,但是 rdds 本身是分布的 collection 而 Array 是还有一个 collection。因此,如果您将 rdd 添加到数组中,您将拥有 collection of collection 这意味着您应该编写两个循环

changed.append(numbersRDD.map(div_two))

def printing(x):
    print x

for i in changed:
    i.foreach(printing)

你的代码和我的代码之间的主要区别是我在将 rdd 添加到已更改的变量时使用了 map(这是一种转换)而不是 foreach(这是一种操作) .我使用了两个循环来打印 rdd

的元素