加入两个 rdds 做一个邻接表

Question

我是 spark 新手，我想加入两个 rdds 来制作邻接列表

RDD1 (nodes): (a, b, c, d, e, f, g)

RDD2 (Edges): ((a,b), (a,e), (f, a), (k,l) ...)

现在我想加入这两个rdds来创建一个这样的邻接列表

( (a,(b,e,..)), (b,(f,....), (g()) ,...)
#assuming that g is not connected to any node , also filter (k,l) because k and l are not in the nodes rdd

稍后我需要找到节点和边的总数。

Answer 1

所以如果我理解正确，您希望有一个邻接列表，其中最终的 RDD 由键值对组成，键是节点，值是它的边缘列表。也许如下所示的是您的想法？尽管我相信如果您希望在最终的 RDD 中显示“g”，那么将它作为 ('g', '') 放在边缘列表中是有意义的，因为您希望传达它没有边缘。

要加入，我们需要将节点列表转换为对 RDD，因此首先我们并行化以创建 RDD，然后映射一个虚拟值，以便我们拥有键值对。

现在我们可以将两个 RDD 相互连接起来，结果将只是两个 RDD 中存在的键，在这种情况下是 'a' 和 'f'。最后，我们剥离我们添加到节点 RDD 和 groupByKey 的虚拟值，以将我们的值组合在一起。

nodes = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
edges = [('a','b'), ('a','e'), ('f', 'a'), ('k','l')]
nodesRDD = sc.parallelize(nodes).map(lambda n: (n, ''))
edgesRDD = sc.parallelize(edges)
joinedRDD = nodesRDD.join(edgesRDD).map(lambda tup: (tup[0], tup[1][1]))
groupedRDD = joinedRDD.groupByKey()

groupedRDD.map(lambda x : (x[0], list(x[1]))).collect()

输出[146]: [('f', ['a']), ('a', ['b', 'e'])]

Count 类似，但现在我们不关心实际的节点值，只关心它们的计数：

nodes = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
edges = [('a','b'), ('a','e'), ('f', 'a'), ('k','l')]
nodesRDD = sc.parallelize(nodes).map(lambda n: (n, 0))
edgesRDD = sc.parallelize(edges).map(lambda tup: (tup[0], 1))
joinedRDD = nodesRDD.join(edgesRDD).map(lambda tup: (tup[0], tup[1][1]))
reducedRDD = joinedRDD.reduceByKey(lambda a, b: a + b)

reducedRDD.collect()

输出[159]: [('f', 1), ('a', 2)]