pyspark将两个rdd合并到一起

Question

我有两个rdd，都是groupby的结果，看起来像：

[(u'1', [u'0']), (u'3', [u'1']), (u'2', [u'0']), (u'4', [u'1'])]

以及

[(u'1', [u'3', u'4']), (u'0', [u'1', u'2'])]

如何将两者合并并得到以下结果：

[(u'1', [u'0',u'3', u'4']]), (u'3', [u'1']), (u'2', [u'0']), (u'4', [u'1']),(u'0', [u'1', u'2'])]

我尝试了join命令，但是没有得到我想要的结果。任何帮助都非常感谢。

Answer 1

我用以下方法解决了这个问题：

rdd2.union(rdd1).reduceByKey(lambda x,y : x+y)

以下这些对我都不起作用：

(rdd1 union rdd2).reduceByKey(_ ++ _)

或者

rdd1.join(rdd2).map(case (k, (ls, rs)) => (k, ls ++ rs))

祝大家好运。

Answer 2

data1 = [(u'1', [u'0']), (u'3', [u'1']), (u'2', [u'0']), (u'4', [u'1'])]
data2 = [(u'1', [u'3', u'4']), (u'0', [u'1', u'2'])]

distData1 = sc.parallelize(data1)
distData2 = sc.parallelize(data2)
distData3 = distData1.leftOuterJoin(distData2)
distData4 = distData3.map(lambda rec : ( rec[0], rec[1][0] + [ ] if rec[1][1] is None else rec[1][1])