Вложенный список Python (Pyspark) reduByKey, список Python добавляется для создания вложенного списка

Question

У меня есть вход RDD в следующем формате:

[('2002', ['cougar', 1]),
('2002', ['the', 10]),
('2002', ['network', 4]),
('2002', ['is', 1]),
('2002', ['database', 13])]

Ключ «2002».Итак, у меня есть пары ключ-значение:

 ('year', ['word', count])

Счетчик - целое число, я хотел бы использовать reduByKey, чтобы получить следующий результат:

[('2002, [['cougar', 1], ['the', 10], ['network', 4], ['is', 1], ['database', 13]]')]

Я изо всех сил пытаюсь получитьсписок гнезд, как указано выше.Основной проблемой является получение вложенного списка.Например, у меня есть три списка a, b и c

a = ['cougar', 1]
b = ['the', 10]
c = ['network', 4]

a.append(b)

вернет a как

 ['cougar', 1, ['the', 10]]

, а

x = []
x.append(a)
x.append(b)

вернет x как

  [['cougar', 1], ['the', 10]]

Однако, если тогда

  c.append(x)

вернет c как

  ['network', 4, [['cougar', 1], ['the', 10]]]

Все вышеперечисленные операции не дают мне желаемого результата.

Я хочу получить

   [('2002', [[word1, c1],[word2, c2], [word3, c3], ...]), 
   ('2003'[[w1, count1],[w2, count2], [w3, count3], ...])]

то есть вложенный список должен быть:

  [a, b, c]

Где a, b, c сами являются списком с двумя элементами.

Надеюсь, вопрос понятен и есть какой-нибудь совет?

Answer 1

Нет необходимости использовать ReduceByKey для этой проблемы.

• Определить СДР

rdd = sc.parallelize([('2002', ['cougar', 1]),('2002', ['the', 10]),('2002', ['network', 4]),('2002', ['is', 1]),('2002', ['database', 13])])

• См. Значения СДРс rdd.collect():

[('2002', ['cougar', 1]), ('2002', ['the', 10]), ('2002', ['network', 4]), ('2002', ['is', 1]), ('2002', ['database', 13])]

• Примените функцию groupByKey и отобразите значения в виде списка, как вы можете видеть в документах Apache Spark .

rdd_nested = rdd.groupByKey().mapValues(list)

• См. Сгруппированные значения СДР rdd_nested.collect():

[('2002', [['cougar', 1], ['the', 10], ['network', 4], ['is', 1], ['database', 13]])]

Answer 2

У меня есть одно решение:

def wagg(a,b):  
    if type(a[0]) == list: 
        if type(b[0]) == list:
            a.extend(b)
        else: 
            a.append(b)
        w = a
    elif type(b[0]) == list: 
        if type(a[0]) == list:
            b.extend(a)
        else:    
            b.append(a)
        w = b
    else: 
        w = []
        w.append(a)
        w.append(b)
    return w  


rdd2 = rdd1.reduceByKey(lambda a,b: wagg(a,b)) 

У кого-нибудь есть лучшее решение?