基于Hadoop實(shí)現(xiàn)Knn算法

更新時(shí)間：2018年12月26日 14:11:01 作者：Angelababy_huan

這篇文章主要為大家詳細(xì) 介紹了基于Hadoop實(shí)現(xiàn)Knn算法的相關(guān)資料，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

Knn算法的核心思想是如果一個(gè)樣本在特征空間中的K個(gè)最相鄰的樣本中的大多數(shù)屬于某一個(gè)類(lèi)別，則該樣本也屬于這個(gè)類(lèi)別，并具有這個(gè)類(lèi)別上樣本的特性。該方法在確定分類(lèi)決策上只依據(jù)最鄰近的一個(gè)或者幾個(gè)樣本的類(lèi)別來(lái)決定待分樣本所屬的類(lèi)別。Knn方法在類(lèi)別決策時(shí)，只與極少量的相鄰樣本有關(guān)。由于Knn方法主要靠周?chē)邢薜泥徑臉颖?，而不是靠判別類(lèi)域的方法來(lái)確定所屬類(lèi)別的，因此對(duì)于類(lèi)域的交叉或重疊較多的待分樣本集來(lái)說(shuō)，Knn方法較其他方法更為合適。

Knn算法流程如下：

1. 計(jì)算當(dāng)前測(cè)試數(shù)據(jù)與訓(xùn)練數(shù)據(jù)中的每條數(shù)據(jù)的距離

2. 圈定距離最近的K個(gè)訓(xùn)練對(duì)象，作為測(cè)試對(duì)象的近鄰

3. 計(jì)算這K個(gè)訓(xùn)練對(duì)象中出現(xiàn)最多的那個(gè)類(lèi)別，并將這個(gè)類(lèi)別作為當(dāng)前測(cè)試數(shù)據(jù)的類(lèi)別

以上流程是Knn的大致流程，按照這個(gè)流程實(shí)現(xiàn)的MR效率并不高，可以在這之上進(jìn)行優(yōu)化。在這里只寫(xiě)，跟著這個(gè)流程走的MR實(shí)現(xiàn)過(guò)程。

Mapper的設(shè)計(jì)：

由于測(cè)試數(shù)據(jù)相比于訓(xùn)練數(shù)據(jù)來(lái)說(shuō)，會(huì)小很多，因此將測(cè)試數(shù)據(jù)用Java API讀取，放到內(nèi)存中。所以，在setup中需要對(duì)測(cè)試數(shù)據(jù)進(jìn)行初始化。在map中，計(jì)算當(dāng)前測(cè)試數(shù)據(jù)與每條訓(xùn)練數(shù)據(jù)的距離，Mapper的值類(lèi)型為：<Object, Text, IntWritable,MyWritable>。map輸出鍵類(lèi)型為IntWritable，存放當(dāng)前測(cè)試數(shù)據(jù)的下標(biāo)，輸出值類(lèi)型為MyWritable，這是自定義值類(lèi)型，其中存放的是距離以及與測(cè)試數(shù)據(jù)比較的訓(xùn)練數(shù)據(jù)的類(lèi)別。

public class KnnMapper extends Mapper<Object, Text, IntWritable,MyWritable> {
 Logger log = LoggerFactory.getLogger(KnnMapper.class);
 private List<float[]> testData;
 @Override
 protected void setup(Context context)
 throws IOException, InterruptedException {
 // TODO Auto-generated method stub
 Configuration conf= context.getConfiguration();
 conf.set("fs.defaultFS", "master:8020");
 String testPath= conf.get("TestFilePath");
 Path testDataPath= new Path(testPath);
 FileSystem fs = FileSystem.get(conf);
 this.testData = readTestData(fs,testDataPath);
 }
 
 @Override
 protected void map(Object key, Text value, Context context)
 throws IOException, InterruptedException {
 // TODO Auto-generated method stub
 String[] line = value.toString().split(",");
 float[] trainData = new float[line.length-1];
 for(int i=0;i<trainData.length;i++){
 trainData[i] = Float.valueOf(line[i]);
 log.info("訓(xùn)練數(shù)據(jù)："+line[i]+"類(lèi)別："+line[line.length-1]);
 }
 for(int i=0; i< this.testData.size();i++){
 float[] testI = this.testData.get(i);
 float distance = Outh(testI, trainData);
 log.info("距離："+distance);
 context.write(new IntWritable(i), new MyWritable(distance, line[line.length-1]));
 }
 }
 
 
 private List<float[]> readTestData(FileSystem fs,Path Path) throws IOException {
 //補(bǔ)充代碼完整
 FSDataInputStream data = fs.open(Path);
 BufferedReader bf = new BufferedReader(new InputStreamReader(data));
 String line = "";
 List<float[]> list = new ArrayList<>();
 while ((line = bf.readLine()) != null) {
 String[] items = line.split(",");
 float[] item = new float[items.length];
 for(int i=0;i<items.length;i++){
 item[i] = Float.valueOf(items[i]);
 }
 list.add(item);
 }
 return list;
 }
 // 計(jì)算歐式距離
 private static float Outh(float[] testData, float[] inData) {
 float distance =0.0f;
 for(int i=0;i<testData.length;i++){
 distance += (testData[i]-inData[i])*(testData[i]-inData[i]);
 }
 distance = (float)Math.sqrt(distance);
 return distance;
 }
}

自定義值類(lèi)型MyWritable如下：

public class MyWritable implements Writable{
 private float distance;
 private String label;
 public MyWritable() {
 // TODO Auto-generated constructor stub
 }
 public MyWritable(float distance, String label){
 this.distance = distance;
 this.label = label;
 }
 @Override
 public String toString() {
 // TODO Auto-generated method stub
 return this.distance+","+this.label;
 }
 @Override
 public void write(DataOutput out) throws IOException {
 // TODO Auto-generated method stub
 out.writeFloat(distance);
 out.writeUTF(label);
 }
 @Override
 public void readFields(DataInput in) throws IOException {
 // TODO Auto-generated method stub
 this.distance = in.readFloat();
 this.label = in.readUTF();
 
 }
 public float getDistance() {
 return distance;
 }
 
 public void setDistance(float distance) {
 this.distance = distance;
 }
 
 public String getLabel() {
 return label;
 }
 
 public void setLabel(String label) {
 this.label = label;
 }
 
}

在Reducer端中，需要初始化參數(shù)K，也就是圈定距離最近的K個(gè)對(duì)象的K值。在reduce中需要對(duì)距離按照從小到大的距離排序，然后選取前K條數(shù)據(jù)，再計(jì)算這K條數(shù)據(jù)中，出現(xiàn)次數(shù)最多的那個(gè)類(lèi)別并將這個(gè)類(lèi)別與測(cè)試數(shù)據(jù)的下標(biāo)相對(duì)應(yīng)并以K，V的形式輸出到HDFS上。

public class KnnReducer extends Reducer<IntWritable, MyWritable, IntWritable, Text> {
 private int K;
 @Override
 protected void setup(Context context)
 throws IOException, InterruptedException {
 // TODO Auto-generated method stub
 this.K = context.getConfiguration().getInt("K", 5);
 }
 @Override
 /***
 * key => 0
 * values =>([1,lable1],[2,lable2],[3,label2],[2.5,lable2])
 */
 protected void reduce(IntWritable key, Iterable<MyWritable> values,
 Context context) throws IOException, InterruptedException {
 // TODO Auto-generated method stub
 MyWritable[] mywrit = new MyWritable[K];
 for(int i=0;i<K;i++){
 mywrit[i] = new MyWritable(Float.MAX_VALUE, "-1");
 }
 // 找出距離最小的前k個(gè)
 for (MyWritable m : values) {
 float distance = m.getDistance();
 String label = m.getLabel();
 for(MyWritable m1: mywrit){
 if (distance < m1.getDistance()){
  m1.setDistance(distance);
  m1.setLabel(label);
 }
 }
 }
 // 找出前k個(gè)中，出現(xiàn)次數(shù)最多的類(lèi)別
 String[] testClass = new String[K];
 for(int i=0;i<K;i++){
 testClass[i] = mywrit[i].getLabel();
 }
 String countMost = mostEle(testClass);
 context.write(key, new Text(countMost));
 }
 public static String mostEle(String[] strArray) { 
  HashMap<String, Integer> map = new HashMap<>(); 
  for (int i = 0; i < strArray.length; i++) {
 String str = strArray[i];
   if (map.containsKey(str)) {
 int tmp = map.get(str);
 map.put(str, tmp+1);
 }else{
 map.put(str, 1);
 }
 }
  // 得到hashmap中值最大的鍵，也就是出現(xiàn)次數(shù)最多的類(lèi)別
  Collection<Integer> count = map.values();
  int maxCount = Collections.max(count);
  String maxString = "";
  for(Map.Entry<String, Integer> entry: map.entrySet()){
   if (maxCount == entry.getValue()) {
 maxString = entry.getKey();
 }
  }
  return maxString; 
 }
}

最后輸出結(jié)果如下：