spark 機器學習 knn 代碼實現(二)

經過knn 算法規則，計算出s2表中的員工所屬的類別
原始數據：
某公司工資表 s1（訓練數據）
格式：員工ID，員工類別，工做年限，月薪（K爲單位）
       101       a類       8年    30k
[hadoop@h201 sss]$ cat s1.txt
101,a,8,30
102,a,6,35
103,a,12,42
104,b,1,6
105,b,1,5
106,a,3,50

沒有分類的 員工工資表 s2（測試數據）
格式：員工ID,  工做年限，  月薪
       108      1年        3.5k            
[hadoop@h201 sss]$ cat s2.txt
108,1,3.5
109,6,22

如下代碼爲了方便初學者學習和理解，我把代碼分開步驟展現，若是有spark開發經驗能夠把代碼合併爲spark腳本，或方法重寫，可以減小上面代碼中的冗餘。
1.初始數據
1.1
scala> val train1=sc.textFile("hdfs://h201:9000/s1.txt")
//樣本數據
scala> val test1=sc.textFile("hdfs://h201:9000/s2.txt")
//測試數據
1.2
scala> val cart1=test1 cartesian train1
//笛卡爾積
scala> cart1.collect
 Array[(String, String)] = Array((108,1,3.5,101,a,8,30), (108,1,3.5,102,a,6,35), (108,1,3.5,103,a,12,42).....
1.3
val c1=cart1.map(_.toString()).map(a=>{
   val a1=a.split(",")
   val aa1=a1(0).replaceAll("\\(","")
   val aa2=a1(1)
   val aa3=a1(2)
   val aa4=a1(3)
   val aa5=a1(4)
   val aa6=a1(5)
   val aa7=a1(6).replaceAll("\\)","")
   (aa1,(aa2,aa3,aa4,aa5,aa6,aa7))
})
//轉換爲key,value結構數據

2.1 歐式距離
def eur(x1: Double,y1: Double,x2: Double,y2: Double): Double = {
         val d1=Math.sqrt(Math.pow(x1 - x2, 2) + Math.pow(y1 - y2, 2))
         return d1
}
//math.pow 算數.平方 ，math.sqrt  算數.開根
eur(1,3.5,8,30,102)

val c2 =c1.groupByKey.flatMap(line =>{
  val h1 = line._2.toArray.map{case(x1,y1,bid,fenlei,x2,y2)=>(line._1,Math.floor(eur(x1.toDouble,y1.toDouble,x2.toDouble,y2.toDouble)),fenlei,bid)}
  (h1)
})
//每一個新加入的數據 距離訓練數據的距離
//Math.floor 做用去除小數位

2.2肯定k值（k=3）
val c3=c2.map(a=>{
       val a11=a._1
       val a22=(a._2,a._3,a._4)
        (a11,a22)
    }).groupByKey().map(b=>{
      val b1=b._1
      val b2=b._2.toArray.sortBy(x=>x._1).take(3)
      (b1,b2)
 })
//sortBy   reverse參數 爲scala語言中，array排序方法的降序表達,不加爲升序表達
//spark RDD中 sortBy(x=>x._1,false,1)  :false 爲降序排列，1爲分區數

3.K點中出現次數最多的分類（肯定分類）
val c4=c3.map(a=>{
   val a1=a._1
   val a2=a._2.map(b=>b._2)
   (a1,a2)
 })


val c5=c4.flatMap(line=>{
    val u1=line._2.map(a=>((line._1+"@"+a).toString,1))
   (u1)
 }).reduceByKey(_+_)

c5.sortBy(a=>a._2,false).take(2)
結果：

員工ID：108 屬於b類
員工ID：109 屬於a類