Max values for each key RDD

Admin 1
Viewed 52

I have this data and want to get each key's max value. The key will be the first element (9,14,26).

(('14', '51600', 'Fashion Week'), 1)
(('9', '61577', 'Guiding Light'), 7)
(('9', '6856', 'Adlina Marie'), 22)
(('14', '120850', 'People Say (feat. Redman)'), 5)
(('26', '155571', "Thinking 'Bout You"), 30)
(('26', '156532', "Hello"), 8)

The final format will be:

 (9, '6856', 'Adlina Marie', 22)
(14, '120850', 'People Say (feat. Redman)', 5)
(26, '155571', "Thinking 'Bout You", 30)

How to select the first column as the key and the last as the value to find the maximum of the value? I tried

groupbykey(lambda x:int(x[0][0])).mapValues(lambda x: max(x))

but it takes the second column as the value to find the max.

rdd pyspark apache-spark max key
2 Answers

You could use map before the aggregations and after:

rdd = rdd.map(lambda x: (x[0][0],(x[1], x[0][1], x[0][2])))
rdd = rdd.groupByKey().mapValues(max)
rdd = rdd.map(lambda x: (x[0], x[1][1], x[1][2], x[1][0]))

Full example:

sc = spark.sparkContext
data = [(('14', '51600', 'Fashion Week'), 1),
        (('9', '61577', 'Guiding Light'), 7),
        (('9', '6856', 'Adlina Marie'), 22),
        (('14', '120850', 'People Say (feat. Redman)'), 5),
        (('26', '155571', "Thinking 'Bout You"), 30),
        (('26', '156532', "Hello"), 8)]
rdd = sc.parallelize(data)

rdd = rdd.map(lambda x: (x[0][0],(x[1], x[0][1], x[0][2])))
print(rdd.collect())
# [('14', (1, '51600', 'Fashion Week')), ('9', (7, '61577', 'Guiding Light')), ('9', (22, '6856', 'Adlina Marie')), ('14', (5, '120850', 'People Say (feat. Redman)')), ('26', (30, '155571', "Thinking 'Bout You")), ('26', (8, '156532', 'Hello'))]

rdd = rdd.groupByKey().mapValues(max)
print(rdd.collect())
# [('14', (5, '120850', 'People Say (feat. Redman)')), ('9', (22, '6856', 'Adlina Marie')), ('26', (30, '155571', "Thinking 'Bout You"))]

rdd = rdd.map(lambda x: (x[0], x[1][1], x[1][2], x[1][0]))
print(rdd.collect())
# [('14', '120850', 'People Say (feat. Redman)', 5), ('9', '6856', 'Adlina Marie', 22), ('26', '155571', "Thinking 'Bout You", 30)]
user_27535010

If working with rdds is not a restriction, here is another approach using a spark df with a window function:

df = spark.createDataFrame(
    [
    (('14', '51600', 'Fashion Week'), 1)
    ,(('9', '61577', 'Guiding Light'), 7)
    ,(('9', '6856', 'Adlina Marie'), 22)
    ,(('14', '120850', 'People Say (feat. Redman)'), 5)
    ,(('26', '155571', "Thinking 'Bout You"), 30)
    ,(('26', '156532', "Hello"), 8)
    ],['key','value']
)

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

df\
    .select(F.col('key._1').alias('key_1'),
            F.col('key._2').alias('key_2'),
            F.col('key._3').alias('key_3'),
            F.col('value'))\
    .withColumn('max', F.max(F.col('value')).over(Window.partitionBy('key_1')))\
    .filter(F.col('value')==F.col('max'))\
    .select('key_1', 'key_2', 'key_3', 'value')\
    .show()

+-----+------+--------------------+-----+
|key_1| key_2|               key_3|value|
+-----+------+--------------------+-----+
|   14|120850|People Say (feat....|    5|
|   26|155571|  Thinking 'Bout You|   30|
|    9|  6856|        Adlina Marie|   22|
+-----+------+--------------------+-----+
user_116985620
Related