PySpark ReduceByKey

Question

Я пытался заставить его работать какое-то время, но каждый раз терпел неудачу. У меня есть 2 файла. Один имеет список имен:

Name1 
Name2 
Name3 
Name4 

Другой список значений, связанные с именами на каждый день в течение года в течение нескольких лет:

['0.1,0.2,0.3,0.4', 
'0.5,0.6,0.7,0.8', 
'10,1000,0.2,5000' 
    ...] 

Цель состоит в том, чтобы иметь выход как:

Name1: [0.1,0.5,10] 
Name2: [0.2,0.6,1000] 
Name3:[0.3,0.7,0.2] 
Name4:[0.4,0.8,5000] 

И затем график графика для каждого. Я написал картограф, который создает список кортежей, который производит следующий вывод (это объект РДДА):

[[('Name1', [0.1]),('Name2', [0,2]),('Name3', [0.3]),('Name4', [0.4])], 
[('Name1', [0.5]),('Name2', [0,6]),('Name3', [0.7]),('Name4', [0.8])], 
[('Name1', [10]),('Name2', [1000]),('Name3', [0.8]),('Name4', [5000])]] 

Теперь нужно объединить все значения для каждого имени в одном списке, но каждую карту с помощью ключа , значение, которое я попытался, возвращает неверный результат.

Answer 1

Вы можете просто прокрутить каждый из них и создать словарь из него с помощью dict.setdefault(). Пример -

>>> ll = [[('Name1', [0.1]),('Name2', [0,2]),('Name3', [0.3]),('Name4', [0.4])], 
... [('Name1', [0.5]),('Name2', [0,6]),('Name3', [0.7]),('Name4', [0.8])], 
... [('Name1', [10]),('Name2', [1000]),('Name3', [0.8]),('Name4', [5000])]] 
>>> d = {} 
>>> for i in ll: 
...  for tup in i: 
...    d.setdefault(tup[0],[]).extend(tup[1]) 
... 
>>> pprint.pprint(d) 
{'Name1': [0.1, 0.5, 10], 
'Name2': [0, 2, 0, 6, 1000], 
'Name3': [0.3, 0.7, 0.8], 
'Name4': [0.4, 0.8, 5000]} 

---

Для Pyspark RDD объекта, попытайтесь просто уменьшить функцию, такие как -

func = lambda x,y: x+y 

Затем отправить это к reduceByKey методе -

object.reduceByKey(func) 

---

По комментариям, на самом деле OP имеет список RDD Ob ject (не один объект RDD), в этом случае вы можете преобразовать объекты RDD в список, вызывая .collect(), а затем выполните логику, а затем вы можете решить, хотите ли вы, чтобы результирующий в качестве словаря python или объекта RDD, если вы хотите сначала. Вы можете позвонить dict.items(), чтобы получить пары ключ-значение и позвонить sc.parrallelize. Пример -

d = {} 
for i in ll: 
    c = i.collect() 
    for tup in i: 
      d.setdefault(tup[0],[]).extend(tup[1]) 

rddobj = sc.parallelize(d.items())