对RDD对键和值进行分组和合并

Question

RDD_Input =  [(('377', '80'), ('1', '4')), (('377', '510'), ('1', '5')), (('377', '79'), ('1', '4')), (('377', '791'), ('1', '1')), (('377', '511'), ('1', '4')), (('377', '433'), ('1', '3')), (('377', '687'), ('1', '1')), (('377', '456'), ('1', '1')), (('377', '399'), ('1', '4')), (('377', '96'), ('1', '5')), (('377', '780'), ('1', '1')), (('377', '683'), ('1', '1')), (('377', '403'), ('1', '5')), (('377', '999'), ('1', '4')), (('377', '502'), ('1', '4')), (('377', '435'), ('1', '5')), (('377', '550'), ('1', '5')), (('377', '948'), ('1', '1')), (('377', '393'), ('1', '4')), (('377', '648'), ('1', '4'))]

输入RDD在键值对((movie1, movie2), (rating1, rating2))中。

如何将RDD转换为((movie1, movie2), (rating1, rating2), (rating3, rating4), (rating5, rating6), ...))？

预期结果示例：(('377', '399'), ('1', '4'), ('1', '4'))

作为关键字的CCD_ 4，并且基于相同的关键字附加以下元组索引。

要求是使用纯粹的RDD API。

Answer 1

看起来，您希望首先将值放入列表中的groupByKey，然后只放入map。

data = [(('a', 'b'), ('1', '4')), (('a', 'b'), ('3', '5')), (('c', 'd'), ('2', '2'))]
rdd = sc.parallelize(data)
rdd = rdd.groupByKey().mapValues(list)
rdd = rdd.map(lambda x: (x[0], *x[1]))
print(rdd.collect())
# [(('c', 'd'), ('2', '2')), (('a', 'b'), ('1', '4'), ('3', '5'))]