如何在Pyspark中通过键找到两个RDD的交集？

Question

这里有两个 RDD：

rdd1 = sc.parallelize([("www.page1.html", "word1"), ("www.page2.html", "word1"), 
    ("www.page1.html", "word3")])

rdd2 = sc.parallelize([("www.page1.html", 7.3), ("www.page2.html", 1.25), 
    ("www.page3.html", 5.41)])

intersection_rdd = rdd1.keys().intersection(rdd2.keys())       

// 当我这样做时，我只得到了键的交集，即（www.page1.html，www.page2.html）。

但是我需要两个rdd中的键以及它们的值。 输出应该像这样：

[www.page1.html, (word1, word3, 7.3)]

[www.page2.html, (word1, 1.25)]

- anvesh

Answer 1

你可以例如使用 cogroup 和过滤器：

## This depends on empty resultiterable.ResultIterable
## evaluating to False

intersection_rdd = rdd1.cogroup(rdd2).filter(lambda x: x[1][0] and x[1][1])
intersection_rdd.map(lambda x: (x[0], (list(x[1][0]), list(x[1][1])))).collect()

## [('www.page1.html', (['word1', 'word3'], [7.3])),
##  ('www.page2.html', (['word1'], [1.25]))]

- zero323

Answer 2

由于您仅针对键使用集合操作，因此您的输出仅包括键。

rdd1.union(rdd2).groupByKey().mapValues(tuple).collect()

Union：将多个PCollection合并为一个PCollection。

GroupByKey：按键对元素进行分组，返回一个键为每个唯一键和值为该键的所有元素的PCollection。

---

('www.page1.html', 'word1')                       ('www.page1.html', ['word1', 'word3', 7.3])

('www.page2.html', 'word1')                       ('www.page2.html', ['word1', 1.25])

('www.page1.html', 'word3')                       ('www.page3.html', [5.41])

('www.page1.html', 7.3)

('www.page2.html', 1.25)

('www.page3.html', 5.41)

- alwaysprep

---

如果您能解释一下为什么原帖中的尝试解决方案是错误的，以及您的解决方案如何解决这个问题，那将非常有帮助。 - Cody Guldner