使用spark求共同好友：

在做一些專案時，又時可能遇到一些類似與求共同好友的要求，可以根據共同好友進行推薦新增好友，就比如說A和B的好友有M,K,O,L，那麼有可能A和B就是也認識，所以可以把B推薦給A，下面是用Spark求共同好友的實現

資料格式如下
A:B,D,E,H,I,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:B,C,D,E,O,M

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
 


object CommFriend {
  def main(args: Array[String]): Unit = {
    var Array(inputPath,outputPath) = args
    inputPath = "D:\\tools\\qianfeng\\hadoop\\Mapreduce\\day09Mapreduce\\fr.txt"
    val conf = new SparkConf().setAppName(s"${this.getClass.getName}").setMaster("local")
    val sc = new SparkContext
 
(conf)
    val lines = sc.textFile(inputPath)
    val friendAndPerson: RDD[(String, List[String])] = lines.flatMap(m => {
      val sp = m.split(":")
      val str: String = sp(0)
      val sp1= sp(1).split(",")
      val tuples: Array[(String, List[String])] = sp1.map(t => {
        //val t: String = t
 

        //一個好友對應一個人
        (t,List(str))
      })
      tuples
    })
  


    //(friend,persons)  把好友相同的人進行聚合
    val friAndPers: RDD[(String, List[String])] = friendAndPerson.reduceByKey((list1, list2) => {
      list1 ::: list2
    })
   
    //每兩個人之間的共同好友
    val ppAndcomFri: RDD[(String, String)] = friAndPers.flatMap(t => {
      val sortedFriend: List[String] = t._2.sortBy(x => x) //這樣可以避免重複出現 A-B 的好友是C ,和 B-A 的好友是C一樣
      val friend: String = t._1 //共同好友
      var commFriends: List[(String, String)] = List[(String, String)]()
      for (i <- 0 until sortedFriend.length) {  //自己和自己不能是好友，且不能重複
        /*
        E  List（A, B ,C） ABC 都有共同好友是E,可以得到 （A-B，E）（A-C，E）（B-C，E）
        對於 B List(N) 代表 只有N有好友B,所以這種就被滿足迴圈條件
         */
        for (j <- i + 1 until sortedFriend.length) {
          //誰跟誰是共同好友（xx-nn,mm） xx 和 nn 的共同好友是 mm
          val tuple: (String, String) = (sortedFriend(i) + "-" + sortedFriend(j), friend)
          commFriends = commFriends ::: List(tuple)
        }
      }
      commFriends
    })

  
    // a-b 的共同好友有n m d h f 。。。
    val gbk: RDD[(String, Iterable[String])] = ppAndcomFri.groupByKey()
    println(gbk)
    gbk.map(m=>{
      (m._1, m._2.mkString(","))
    })
    .coalesce(1).saveAsTextFile(outputPath)

  }

}