时间戳

首页 > 解决方案 > 对pyspark中的值进行排序的RDD操作

python - 对pyspark中的值进行排序的RDD操作

问题描述

我的文件格式如下，

0, Alpha,-3.9, 4, 2001-02-01, 5, 20
0, Beta,-3.8, 3, 2002-02-01, 6, 21
1, Gamma,-3.7, 8, 2003-02-01, 7, 22
0, Alpha,-3.5, 4, 2004-02-01, 8, 23
0, Alpha,-3.9, 4, 2005-02-01, 8, 27

我想通过使用 rdd 操作1st elements对每一行进行排序。3rd elements我更喜欢得到以下输出，

(Beta, 3)
(Alpha, 4)
(Gamma, 8)

这就是我目前所做的，

rdd = sc.textFile(myDataset)
list_ = rdd.map(lambda line: line.split(",")).map(lambda e : e[1]).distinct().collect() 
new_ = list_.sortBy(lambda e : e[2])

但我无法按我的意愿排序。谁能告诉如何仅基于 rdd 的操作来执行此操作？

标签： pythonpython-3.xapache-sparkpysparkrdd

解决方案

rdd = sc.textFile(myDataset)是正确的。

list_ = rdd.map(lambda line: line.split(",")).map(lambda e : e[1]).distinct().collect() 
new_ = list_.sortBy(lambda e : e[2]) # e[2] does not exist.

您已经调用了 collect on list_，因此它不再是 RDD。然后你继续调用sortBy它，所以它不起作用。也许您在发布时犯了这个错误。主要问题是地图操作。您需要创建一个pairWiseRdd，但您还没有创建一个。因此，没有e[2]可排序的。见下文。

>>> rdd.map(lambda line: line.split(",")).map(lambda e : e[1]).collect()
[' Alpha', ' Beta', ' Gamma', ' Alpha', ' Alpha']

上面将没有您需要使用的值distinct() 相反，您需要这样做

>>> list_ = rdd.map(lambda line: line.split(",")).map(lambda e : (e[1],e[3]))
>>> list_.collect()
[(' Alpha', ' 4'),
 (' Beta', ' 3'),
 (' Gamma', ' 8'),
 (' Alpha', ' 4'),
 (' Alpha', ' 4')]
>>> distinct_rdd = list_.distinct() #making stuff distinct
>>> distinct_rdd.collect()
[(' Alpha', ' 4'), (' Beta', ' 3'), (' Gamma', ' 8')]

现在我们已经创建了pairWiseRdd，我们可以使用每对的第二个值对其进行排序。

>>> sorted_rdd = distinct_rdd.sortBy( lambda x:x[1] )
>>> sorted_rdd.collect()
[(' Beta', ' 3'), (' Alpha', ' 4'), (' Gamma', ' 8')]

推荐阅读