Получить верхние значения на основе составного ключа для каждого раздела в Spark RDD - PullRequest
Новая
       44

Получить верхние значения на основе составного ключа для каждого раздела в Spark RDD

2 голосов
/ 15 октября 2019

Я хочу использовать следующий rdd 

rdd = sc.parallelize([("K1", "e", 9), ("K1", "aaa", 9), ("K1", "ccc", 3), ("K1", "ddd", 9),
("B1", "qwe", 4), ("B1", "rty", 7), ("B1", "iop", 8), ("B1", "zxc", 1)])

, чтобы получить вывод 

[('K1', 'aaa', 9),
 ('K1', 'ddd', 9),
 ('K1', 'e', 9),
 ('B1', 'iop', 8),
 ('B1', 'rty', 7),
 ('B1', 'qwe', 4)]

Я сослался на Получить 3 верхних значения для каждого ключа в СДР в Spark и использовал следующий код 

from heapq import nlargest
rdd.groupBy(
    lambda x: x[0]
).flatMap(
    lambda g: nlargest(3, g[1], key=lambda x: (x[2],x[1]))
).collect()

Однако я могу получить только 

[('K1', 'e', 9),
 ('K1', 'ddd', 9),
 ('K1', 'aaa', 9),
 ('B1', 'iop', 8),
 ('B1', 'qwe', 7),
 ('B1', 'rty', 4)]

Как мне поступить?

Ответы [ 2 ]

1 голос
/ 15 октября 2019

На самом деле это проблема сортировки, но sorting - вычислительно очень дорогой процесс из-за shuffling. Но вы можете попробовать: 

rdd2 = rdd.groupBy(
    lambda x: x[0]
).flatMap(
    lambda g: nlargest(3, g[1], key=lambda x: (x[2],x[1]))
)

rdd2.sortBy(lambda x: x[1], x[2]).collect()
# [('K1', 'aaa', 9), ('K1', 'ddd', 9), ('K1', 'e', 9), ('B1', 'iop', 8), ('B1', 'qwe', 4), ('B1', 'rty', 7)]

Я отсортировал это, используя первое и второе значение кортежей.

Также обратите внимание, что q предшествует r в алфавитном порядке. Таким образом, ваш ожидаемый результат отключен и вводит в заблуждение.

0 голосов
/ 15 октября 2019

Если вы открыты для фрейма данных, вы можете использовать функцию windows с rank

Вдохновленный с здесь 

import pyspark.sql.functions as f
from pyspark.sql import functions as F
from pyspark.sql import SparkSession
from pyspark.sql import Window

spark = SparkSession.builder.appName('test').master("local[*]").getOrCreate()

df = spark.createDataFrame([
    ("K1", "e", 9),
    ("K1", "aaa", 9),
    ("K1", "ccc", 3),
    ("K1", "ddd", 9),
    ("B1", "qwe", 4),
    ("B1", "rty", 7),
    ("B1", "iop", 8),
    ("B1", "zxc", 1)], ['A', 'B', 'C']
    )

w = Window.partitionBy('A').orderBy(df.C.desc())
df.select('*', F.rank().over(w).alias('rank')).filter("rank<4").drop('rank').show()


+---+---+---+
|  A | B | C|
+---+---+---+
| B1 | iop | 8|
| B1 | rty | 7|
| B1 | qwe | 4|
| K1 | e | 9|
| K1 | aaa | 9|
| K1 | ddd | 9|
+---+---+---+
...