Spark实现之 好友推荐

网上好多好推荐之类的博客都是基于二度好友的推荐，下面介绍基于邻接表的好友推荐的spark实现。

输入：

1 2,3,4,5,6,7,8
2 1,3,4,5,7
3 1,2
4 1,2,6
5 1,2
6 1,4
7 1,2
8 1

第一列表示用户，后面的表示用户的所有好友。
需求：找出用户好友的好友且与用户不是好友的，推荐给用户。

思路：1、key为user，value可设计成user有关联关系的人及关联关系的说明，例如：

                    key："1"  value：<"2", "yes"> 表示"1"与"2"已经是好友关系

                    key："2" value：<"3", "1">表示"2"与"3"共同的好友为"1"

            2、对reduce后的key的value做归类，比如把key为"2"的value的第一个值为"3"的value归为一组，这一组value中的第二个值若没有"yes"出现，则把"3"推荐给"2"，若有"yes"出现，说明"2"与"3"已为好友关系。

步骤：1、map：读取每行数据，组合user与每一个好友，两两组合user的好友。

            2、reduce：分析value的第一个值是否时key好友，若不是则可以推荐给key。

package dabook

import org.apache.spark.{SparkConf, SparkContext}
import collection.mutable.{ HashMap, MultiMap, Set }

import scala.collection.mutable

object RecommFriend {

def mymap(x:String): List[(String, (String, String))] ={
var result = List[(String, (String, String))]()
val arr = x.split(" ")
if(arr.length != 2){
return result
}
val host = arr(0).trim //主体
val friends = arr(1).trim
val arr_frinds = friends.split(",") //主体的列表

for (s <- arr_frinds){
result ::= (host, (s, "yes")) //yes表示已为好友关系
}

for(i <- Range(0, arr_frinds.length); j <- Range(i+1, arr_frinds.length)){
result ::= (arr_frinds(i), (arr_frinds(j), host)) //好友列表两两分组，host是他们共同的好友
result ::= (arr_frinds(j), (arr_frinds(i), host))
}
result
}

def myreduce(x:Iterable[(String, String)]): String ={
var result = new String()

var mm = new mutable.HashMap[String, Set[String]] with mutable.MultiMap[String, String]
for(s <- x){
mm.addBinding(s._1, s._2)
}
var sb = new mutable.StringBuilder()
val ks = mm.keys
for(k <- ks){
var flag = true
val values = mm.get(k)
for (v <- values.get){
if(v.equals("yes")){
flag = false
}
}
if(flag){
sb.append("推荐好友:" + k + " 共同的好友：")
sb.append(values.get.mkString(", "))
sb.append(" | ")
}
if(sb.length > 2){
result = sb.substring(0, sb.length-3)
}
else {
result = sb.toString()
}
}
result
}

def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[2]").setAppName("recommfriend")
val sc = new SparkContext(sparkConf)
val input = sc.textFile("/home/xdk/file/friend")
val res = input.flatMap(x=>mymap(x)).groupByKey().map(x=>(x._1, myreduce(x._2)))
res.foreach(println)

//test1()

}

def test(): Unit ={
val ss = "1 2,3,4"
val res = mymap(ss)
println(res.size)
println(res)
}

def test1(): Unit ={
var mmm = new mutable.HashMap[String, Set[String]] with mutable.MultiMap[String, String]
mmm.addBinding("123", "asd")
mmm.addBinding("123", "dsad")
mmm.addBinding("43345", "dasd")

val keys1 = mmm.keys
println(keys1)
for(s <- keys1){
println(s)
}
}

}

输出：

(8,推荐好友:2 共同的好友：1 | 推荐好友:5 共同的好友：1 | 推荐好友:7 共同的好友：1 | 推荐好友:4 共同的好友：1 | 推荐好友:6 共同的好友：1 | 推荐好友:3 共同的好友：1)
(7,推荐好友:5 共同的好友：1, 2 | 推荐好友:8 共同的好友：1 | 推荐好友:4 共同的好友：1, 2 | 推荐好友:6 共同的好友：1 | 推荐好友:3 共同的好友：1, 2)
(4,推荐好友:5 共同的好友：1, 2 | 推荐好友:8 共同的好友：1 | 推荐好友:7 共同的好友：1, 2 | 推荐好友:3 共同的好友：1, 2)
(5,推荐好友:8 共同的好友：1 | 推荐好友:7 共同的好友：1, 2 | 推荐好友:4 共同的好友：1, 2 | 推荐好友:6 共同的好友：1 | 推荐好友:3 共同的好友：1, 2)
(6,推荐好友:2 共同的好友：4, 1 | 推荐好友:5 共同的好友：1 | 推荐好友:8 共同的好友：1 | 推荐好友:7 共同的好友：1 | 推荐好友:3 共同的好友：1)
(3,推荐好友:5 共同的好友：1, 2 | 推荐好友:8 共同的好友：1 | 推荐好友:7 共同的好友：1, 2 | 推荐好友:4 共同的好友：1, 2 | 推荐好友:6 共同的好友：1)
(2,推荐好友:8 共同的好友：1 | 推荐好友:6 共同的好友：4, 1)
(1,)