Блог

Используя Pyspark, я пытаюсь работать с RDD для агрегирования на основе содержимого этого RDD.

Мой RDD в настоящее время выглядит так (очевидно, с гораздо большим количеством данных):

 [([u'User1', u'2'], 1), ([u'User2', u'2'], 1), ([u'User1', u'3'], 1)]
 

Я хочу объединить это в формат:

 User1  5
User2  2
 

Я изо всех сил пытаюсь взаимодействовать с RDD, в частности со списками внутри RDD, чтобы получить доступ к этим данным. Я также должен сохранить это в RDD, а не преобразовывать его в фрейм данных.

Кто-нибудь может показать мне, как это сделать, пожалуйста?

Другое решение, очень похожее на @mck, но немного более понятное, — использовать оператор add вместо другой лямбда-функции:

 from operator import add 

rdd = sc.parallelize([("user1", "2"), ("user2", "2"), ("user1", "3")])
rdd = rdd.map(lambda x: (x[0], int(x[1])))
rdd = rdd.reduceByKey(add)

"""
>>> rdd.collect()
>>> Out[54]: [('user2', 2), ('user1', 5)]
"""
 

Вы можете map преобразовать RDD в формат (user, value) и выполнить reduceByKey группировку по пользователю и суммировать значения.

 result = rdd.map(lambda x: (x[0][0], int(x[0][1]))).reduceByKey(lambda x, y: x   y)

result.collect()
# [('User2', 2), ('User1', 5)]