通过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类