基于Hadoop实现Knn算法

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("训练数据："+line[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 {
 //补充代码完整
 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;
 }
 // 计算欧式距离
 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;
 }
}

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;
 }
 
}

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个
 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个中，出现次数最多的类别
 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中值最大的键，也就是出现次数最多的类别
  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; 
 }
}