在scala中将集合的元素与HashMap的键合并

Question

我希望有一个简单的方法来解决我有两个RDD

g.vertices
(4,Set(5, 3))
(0,Set(1, 4))
(1,Set(2))
(6,Set())
(3,Set(0))
(5,Set(2))
(2,Set(1))

maps
Map(4 -> Set(5, 3))
Map(0 -> Set(1, 4))
Map(1 -> Set(2))
Map(6 -> Set())
Map(3 -> Set(0))
Map(5 -> Set(2))
Map(2 -> Set(1))

我怎么能做这样的事？

(4,Map(5 -> Set(2), 3 -> Set(0)))
(0,Map(1 -> Set(2), 4 -> Set(5, 3)))
(1,Map(2 -> Set(1)))
(6,Map())
(3,Map(0 -> Set(1, 4)))
(5,Map(2 -> Set(1)))
(2,Map(1 -> Set(2)))

我想将贴图的键与集合的元素结合起来，所以我想更改集合的元素（将它们与贴图的键合并）

我想到

val maps = g.vertices.map { case (id, attr) => HashMap(id -> attr) }

g.mapVertices{case (id, data) => data.map{case vId => maps.
  map { case i if i.keySet.contains(vId) => HashMap(vId -> i.values) } }}

但我有一个错误

org.apache.spark.SparkExc0019：RDD转换和操作只能由驱动程序调用，不能在其他转换内部调用；例如，rdd1.map（x=

Answer 1

这是一个用于连接的简单用例。在以下代码中，A是g.vertices中的键的类型，K和V是映射的键和值类型：

def joinByKeys[A, K, V](sets: RDD[(A, Set[K])], maps: RDD[Map[K, V]]): RDD[(A, Map[K, V])] = {
  val flattenSets = sets.flatMap(p => p._2.map(_ -> p._1)) // create a pair for each element of vertice's sets
  val flattenMaps = maps.flatMap(identity)                 // create an RDD with all indexed values in Maps
  flattenMaps.join(flattenSets).map{                       // join them by their key
    case (k, (v, a)) => (a, (k, v))                        // reorder to put the vertexId as id
  }.aggregateByKey(Map.empty[K, V])(_ + _, _ ++ _)         // aggregate the maps

}