迭代查找地图

Question

def description(list:Array[String]): Array[String] = { 
    for (y <- list) yield modulelookup.lookup(take(4)) + " " + brandlookup.lookup(y.drop(4)).toString() 
} 

val printRDD = outputRDD.collect().map(x=> (description(x._1),x._2)) 

是我当前的代码。我想这样做没有收集。模块查找和品牌查找是RDD。这个怎么做？

Answer 1

如果modulelookup和brandlookup都比较小，你可以将这些播放变量和使用映射如下：

val modulelookupBD = sc.broadcast(modulelookup.collectAsMap) 
val brandlookupBD = sc.broadcast(brandlookup.collectAsMap) 

def description(list:Array[String]): Array[String] = list.map(x => { 
    val module = modulelookupBD.value.getOrElse(x.take(4), "") 
    val brand = brandlookupBD.value.getOrElse(x.drop(4), "") 
    s"$module $brand" 
}) 

val printRDD = outputRDD.map{case (xs, y) => (description(xs), y)} 

如果不是有处理这种没有效率的方式。您可以尝试使用flatMap,join和groupByKey，但对于任何大型数据集，此组合可能过于昂贵。

val indexed = outputRDD.zipWithUniqueId 
val flattened = indexed.flatMap{case ((xs, _), id) => xs.map(x => (x, id))} 

val withModuleAndBrand = flattened 
    .map(xid => (xid._1.take(4), xid)) 
    .join(modulelookup) 
    .values 
    .map{case ((x, id), module) => (x.drop(4), (id, module))} 
    .join(brandlookup) 
    .values 
    .map{case ((id, module), brand) => (id, s"$module $brand")} 
    .groupByKey 

val final = withModuleAndBrand.join(
    indexed.map{case ((_, y), id) => (id, y)} 
).values 

使用DataFrame替换RDD可以减少样板代码，但性能仍然存在问题。