如何将查找(广播)RDD(或数据集)访问到其他RDD映射函数中

Question

我是spark和scala的新手,刚开始学习......我在CDH 5.1.3上使用spark 1.0.0

我得到了一个名为dbTableKeyValueMap的广播rdd:RDD [(String,String)],我想使用dbTableKeyValueMap来处理我的文件RDD(每行有300多列).这是代码:

val get = fileRDD.map({x =>
  val tmp = dbTableKeyValueMap.lookup(x)
  tmp
})

在本地运行此挂起和/或在一段时间后出错:

scala.MatchError: null
at org.apache.spark.rdd.PairRDDFunctions.lookup(PairRDDFunctions.scala:571)

我可以理解访问一个RDD里面其他会有问题,如果集合的地点和大小进入图片..对于我采取笛卡尔产品不是选项,因为文件RDD中的记录是巨大的(每行有300+列)...就像我使用分布式缓存在setup方法中加载这个dbTableKeyValueMap并在hadoop java mapreduce代码的MAP中使用,我想在spark map中使用类似的方式...我找不到简单的例子来引用类似的用例...一个我想迭代文件RDD行并在"每一列"上进行一些转换,祝福,查找等以进行进一步处理......或者还有其他任何方式我可以使用dbTableKeyValueMap作为scala集合而不是spark RDD

请帮忙

Answer 1

谢谢....最简单的事情是将查找RDD转换为"scala集合"并且很好!我可以使用任何RDD在内部转换中访问它....

val scalaMap = dbTableKeyValueMap.collectAsMap.toMap
val broadCastLookupMap = sc.broadcast(scalaMap)

val get = fileRDD.map({x =>
  val tmp = broadCastLookupMap.value.get(x).head
  tmp
})

这个简单的解决方案应该记录在早期学习者的某个地方.我花了一些时间来弄清楚...

谢谢您的帮助...