[Spark共同好友查找]

共同好友的概念

      在一个庞大的社交网络中，两个互相认识的朋友之间的也会存在共同好友。在这个庞大的社交网络总，对所有的用户对中找到”共同好友”，这是一个复杂及有趣的事情。假设，U为一个用户及其所有好友的一个集合：｛U1,U2,U3,…Un｝,我们要从每组集合（Ui,Uj）（i != j）找出共同好友关系。

       在如今的大多数社交网络（Facebook,LinkedIn,QQ）都提供了有关的服务，可以帮助帮助你与好友共享消息，图片和视频。有些网站甚至还提供了视频聊天服务，帮助你和好友保持联系和分享生活中的乐趣。下图是QQ共同好友界面。

       根据定义，好友就是你认识的，喜欢的或信任的一个人。例如，你在QQ上有一个好友列表，这个社交网络中好友关系是双向的。如果你是我的好友，那么我也就是你的好友。一般情况下，社交网络会尽可能的预先完成计算来减少请求的处理时间，其中一个常见的请求处理就是共同好友特性。访问QQ的好友查找页面时，就可以看到两个好友之间的共同好友个数。这个列表的信息不会频繁改变，如果每次访问这个页面的信息时都要重新计算，这将会浪费时间和资源。

共同好友输入样本数据

       准备一组记录数据作为输入，每个记录的数据格式如下所示：

       <user>,<friend1> <friend2> … <friendn>

这里的<friend1> <friend2> … <friendn>就是用户<user>的好友，每个用户/好友有唯一用户ID标识。以下是一个简单且完整的测试输入数据实例：

A1,A2 A3 A4 A5 A6 A7
A2,A1 A3 A4
A3,A1 A2 A4 A5
A4,A1 A2 A3
A5,A1 A3 A7
A6,A1
A7,A2

共同好友处理流程

       为了帮助我们更好的理解共同好友的查找，假设每一个用户的好友作为一个键-值对，即用户ID作为一个键，用户好友作为一组值，那么将会有以下的一序列map信息。

map(A1,(A2 A3 A4 A5 A6 A7))将生成：

((A1,A2),List(A2, A3, A4, A5, A6, A7))
((A1,A3),List(A2, A3, A4, A5, A6, A7))
((A1,A4),List(A2, A3, A4, A5, A6, A7))
((A1,A5),List(A2, A3, A4, A5, A6, A7))
((A1,A6),List(A2, A3, A4, A5, A6, A7))
((A1,A7),List(A2, A3, A4, A5, A6, A7))

map(A2,(A1 A3 A4))将生成：

((A1,A2),List(A1, A3, A4))
((A2,A3),List(A1, A3, A4))
((A2,A4),List(A1, A3, A4))

map(A3,(A1 A2 A4 A5))将生成：

((A1,A3),List(A1, A2, A4, A5))
((A2,A3),List(A1, A2, A4, A5))
((A3,A4),List(A1, A2, A4, A5))
((A3,A5),List(A1, A2, A4, A5))

map(A4,(A1 A2 A3))将生成：

((A1,A4),List(A1, A2, A3))
((A2,A4),List(A1, A2, A3))
((A3,A4),List(A1, A2, A3))

map(A5,(A1 A3 A7))将生成：

((A1,A5),List(A1, A3, A7))
((A3,A5),List(A1, A3, A7))
((A5,A7),List(A1, A3, A7))

map(A6,(A1))将生成：

((A1,A6),List(A1))

map(A7,(A2))将生成：

((A2,A7),List(A2))

将上面的map信息按键进行分组，则有：

((A1,A3),(List(A2, A3, A4, A5, A6, A7), List(A1, A2, A4, A5)))
((A2,A3),(List(A1, A3, A4), List(A1, A2, A4, A5)))
((A5,A7),(List(A1, A3, A7)))
((A3,A5),(List(A1, A2, A4, A5), List(A1, A3, A7)))
((A1,A5),(List(A2, A3, A4, A5, A6, A7), List(A1, A3, A7)))
((A2,A7),(List(A2)))
((A3,A4),(List(A1, A2, A4, A5), List(A1, A2, A3)))
((A2,A4),(List(A1, A3, A4), List(A1, A2, A3)))
((A1,A2),(List(A2, A3, A4, A5, A6, A7), List(A1, A3, A4)))
((A1,A7),(List(A2, A3, A4, A5, A6, A7)))
((A1,A4),(List(A2, A3, A4, A5, A6, A7), List(A1, A2, A3)))
((A1,A6),(List(A2, A3, A4, A5, A6, A7), List(A1)))

最后，用户的共同好友将生成：

(A1, A3)  [A4, A5, A2]
(A2, A3)  [A4, A1]
(A5, A7)  []
(A3, A5)  [A1]
(A1, A5)  [A3, A7]
(A2, A7)  []
(A3, A4)  [A2, A1]
(A2, A4)  [A3, A1]
(A1, A2)  [A4, A3]
(A1, A7)  []
(A1, A4)  [A2, A3]
(A1, A6)  []

Spark代码实现

/**
  * Spark共同好友查找
  **/
object FindCommonFriends {
    def main(args: Array[String]): Unit = {
        if (args.size < 2) {
            println("Usage: FindCommonFriends <input-dir> <output-dir>")
            sys.exit(1)
        }
        //输入路径
        val inputPath = args(0)
        //输出路径
        val outputPath = args(1)
        //测试环境使用1个内核处理即可，生产环境中进行修改
        val sparkConf: SparkConf = new SparkConf()
            .setMaster("local[1]")
            .setAppName("FindCommonFriends")
        val sc: SparkContext = SparkContext.getOrCreate(sparkConf)

        val records: RDD[String] = sc.textFile(inputPath)
        //映射两两组合键值对
        val pairs: RDD[((String, String), Seq[String])] = records.flatMap(s => {
            val tokens: Array[String] = s.split(",")
            val person: String = tokens(0)
            val friends: Seq[String] = tokens(1).split("\\s+").toList
            val result: Seq[((String, String), Seq[String])] = for {
                i <- 0 until friends.size
                friend = friends(i)
            } yield {
                if (person < friend)
                    ((person, friend), friends)
                else
                    ((friend, person), friends)
            }
            result
        })

        //共同好友计算
        val commonFriends: RDD[((String, String), immutable.Iterable[String])] = pairs
            .groupByKey()
            .mapValues(iter => {
                val friendCount = for {
                    list <- iter
                    if !list.isEmpty
                    friend <- list
                } yield ((friend, 1))
                friendCount.groupBy(_._1).mapValues(_.unzip._2.sum).filter(_._2 > 1).map(_._1)
            })

        //保存结果
        //commonFriends.saveAsTextFile(outputPath)

        //打印共同好友结果
        val formatedResult = commonFriends.map(
            f => s"(${f._1._1}, ${f._1._2})\t${f._2.mkString("[", ", ", "]")}"
        )

        formatedResult.foreach(println)
        sc.stop()
    }
}

运行结果

(A1, A3)	[A4, A5, A2]
(A2, A3)	[A4, A1]
(A5, A7)	[]
(A3, A5)	[A1]
(A1, A5)	[A3, A7]
(A2, A7)	[]
(A3, A4)	[A2, A1]
(A2, A4)	[A3, A1]
(A1, A2)	[A4, A3]
(A1, A7)	[]
(A1, A4)	[A2, A3]
(A1, A6)	[]

结论

     通过共同好友查找算法，我们可以很方便的计算出一个社交网络中用户的共同好友关系。在上面的测试运行结果中，我们可以看到，用户两两之间的共同好友。其中(A5, A7)，(A2, A7)，(A1, A7)，(A1, A6)之间没有共同好友。