spark

問題描述

在商品推薦的業(yè)務(wù)邏輯計算時，遇到一個計算商品偏好權(quán)重的問題：實時權(quán)重要和離線權(quán)重結(jié)合，其中key相同的商品權(quán)重求和，不相同的保留實時和離線的權(quán)重不變，由于業(yè)務(wù)代碼是實時計算，只能用scala處理（不會java = =|）具體數(shù)據(jù)如下：

時間節(jié)點1：
bean_json:online_rating       value=C01:B02:12.00#C01:B01:8.00
bean_json:real_time_rating    value=C01:B01:40.00:100#C01:B02:60.00:100   
時間節(jié)點2：
bean_json:online_rating       value=C01:B02:9.00#C01:B04:8.00#C01:B01:8.00#C01:B03:6.00                                                 
bean_json:real_time_rating    value=C01:B03:30.00:100#C01:B04:40.00:100#C01:B02:30.00:100 

業(yè)務(wù)計算邏輯

在時間節(jié)點1時
商品的實時權(quán)重為：
C01:B01 = 40.00  
C01:B02 = 60.00
離線權(quán)重 = 實時權(quán)重*衰減系數(shù)0.2
C01:B02 = 60*0.2 = 12.00  
C01:B01 = 40*0.2 = 8.00

時間節(jié)點2時
新的實時權(quán)重為：
C01:B03 = 30.00  
C01:B04 = 40.00 
C01:B02 = 30.00 
新的離線權(quán)重：
C01:B02 = (12.00 + 30*0.2)/2 = 9.00 
C01:B04 = 40*0.2 = 8.00
C01:B01 = 8.00 
C01:B03 = 30*0.2 = 6.00

摸索了很久終于在這篇文章找到了方法：scala 兩個map合并，key相同時value相加

主要原理是將兩個權(quán)重數(shù)據(jù)轉(zhuǎn)成Map[key,value],使用函數(shù)getOrElse將兩個Map合并時，做有條件的處理，條件就是兩個Map的key是否相同，相同就做計算，不相同則直接保留。

一行代碼即可得到結(jié)果，代碼如下：

val unionMap = map1 ++ map2.map(t => t._1 -> (t._2 + map1.getOrElse(t._1, t._2))/2) //key相同權(quán)重求和除2

具體代碼

/**
    * 計算新的權(quán)重
    * @param realTimeWeight 實時權(quán)重
    * @param onlineWeight   離線權(quán)重
    * @return 新權(quán)重
    */
def dealNewOnRating2(realTimeWeight:String,onlineWeight:String): String = {
    var newWeightArray = new StringBuilder

    val realTimeWeightRdd: Array[(String, Double)] = realTimeWeight.split("#").map(w=>(w,1)).map(x =>{
      val realTimeInfo = x._1.split(":")
      val cateInfo = realTimeInfo(0)+":"+realTimeInfo(1)
      val rtWeight = realTimeInfo(2).toDouble * 0.2
      (cateInfo,rtWeight)
    })

    val onlineWeightRdd: Array[(String, Double)] = onlineWeight.split("#").map(w=>(w,1)).map(x =>{
      val realTimeInfo = x._1.split(":")
      val cateInfo = realTimeInfo(0)+":"+realTimeInfo(1)
      val onWeight = realTimeInfo(2).toDouble
      (cateInfo,onWeight)
    })

    val map1 = realTimeWeightRdd.toMap
    val map2 = onlineWeightRdd.toMap
    val unionMap = map1 ++ map2.map(t => t._1 -> (t._2 + map1.getOrElse(t._1, t._2))/2) //key相同權(quán)重求和除2
    val unionWeight = unionMap.toArray.sortBy(_._2)(Ordering[Double].reverse).take(10)  //排序取top10

    for (i <- unionWeight.indices) {
      val cateInfo = unionWeight(i)._1
      val newWeight = unionWeight(i)._2.formatted("%.2f").toString
      newWeightArray ++= cateInfo++=":"++=newWeight++="#"
    }
    newWeightArray.toString
  }