迭代查找地圖

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可以減少樣板代碼，但性能仍然存在問題。