删除元组并创建一个新的排序列表

Question

我有一个 RDD，它是我使用 PySpark 创建的，大小约为 600 GB通过看起来完全像这样的键值加入。

[('43.72_-70.08', (('0744632', -70.08, 43.72, '2.4'), '18090865')),
 ('43.72_-70.08', (('0744632', -70.08, 43.72, '2.4'), '18090865')),
 ('43.25_-67.58', (('0753877', -67.58, 43.25, '7.2'), '18050868')),
 ('43.01_-75.24', (('0750567', -75.24, 43.01, '7.2'), '18042872'))]

我想要这样的东西并按第一个元素排序：

['0744632', '18090865', '2.4',
'0744632', '18090865', '2.4',
'0750567', '18042872', '7.2',
'0753877', '18050868', '7.2']

有没有办法从元组中获取数据并以所需格式输出。

注意：这是一个 600 GB 的 RDD，第一列有超过一百万个不同的值，大约有 100 万个不同的值。 150 亿行，如果可能的话，我将不胜感激。

Answer 1

在您的 spark 集群中执行此操作，例如：

In []:
(rdd.map(lambda x: (x[1][0][0], x[1][1], x[1][0][2]))
 .sortBy(lambda x: x[0])
 .flatMap(lambda x: x)
 .collect())

Out[]:
['0744632', '18090865', 43.72, '0744632', '18090865', 43.72, '0750567', 
 '18042872', 43.01, '0753877', '18050868', 43.25]

或者

In []:
import operator as op

(rdd.map(lambda x: (x[1][0][0], x[1][1], x[1][0][2]))
 .sortBy(lambda x: x[0])
 .reduce(op.add))

Out[]:
('0744632', '18090865', 43.72, '0744632', '18090865', 43.72, '0750567', 
 '18042872', 43.01, '0753877', '18050868', 43.25)

这似乎是一个相当笨拙的结构，如果您指的是元组列表，则只需删除 flatMap():

In []:
(rdd.map(lambda x: (x[1][0][0], x[1][1], x[1][0][2]))
 .sortBy(lambda x: x[0])
 .collect())

Out[]:
[('0744632', '18090865', 43.72),
 ('0744632', '18090865', 43.72),
 ('0750567', '18042872', 43.01),
 ('0753877', '18050868', 43.25)]

Answer 2

这是一个简单的单行解决方案

sorted([(x[1][0][0], x[1][1], x[1][0][3]) for x in your_list])

我认为它比基于此 post 的 lambda 解决方案稍快 What is the difference between these two solutions - lambda or loop - Python

Answer 3

与另一个 Spark 答案相似：

rdd=rdd.map(lambda (a,(b,c)): [b[0], c, b[3]])\
       .sortBy(lambda row: row[0])

你也可以使用 reduce 而不是 flatMap:

rdd.reduce(lambda x,y: x+y)

删除元组并创建一个新的排序列表

Remove the tuple and create a new sorted list

python

tuples

pyspark

spark-dataframe

pyhdfs-client