Hadoop-k_means算法的并行化&非并行化实现

实 验 内 容

自 行 准 备 数 据 集 ， 设 计 一 个 数 据 挖 掘 算 法 （ 聚 类 、 分 类 、 频 繁 项 集 挖 掘 或 其 他主 题 ） 对 数 据 集 进 行 信 息 提 取 ， 要 求 分 别 使 用 并 行 化 和 非 并 行 化 的 方 式 实 现 该 算法 。

实 验 要 求

自 行 对 比 并 行 化 和 非 并 行 化 实 现 方 法 的 数 据 挖 掘 结 果 ， 两 种 结 果 需 完 全 一致 。

有用的参考链接：

UCI数据集

MapReduce Kmeans聚类算法

（java）K-means算法

---

我选择实现的是k_means聚类算法。

首先是数据集，我用的是UCI的wine.data，可以点击查看数据集的信息，右键目标另存为直接down下来。

k_means算法的思想：

1）从N个文档随机选取K个文档作为质心

2）对剩余的每个文档测量其到每个质心的距离，并把它归到最近的质心的类。距离采用标准化欧式距离公式计算。

3）重新计算已经得到的各个类的质心

4）迭代2～3步直至新的质心与原质心相等或小于指定阈值，算法结束

---

并行化实现

combiner可以省略，直接reduce。

---

非并行化实现

思路相同。

数据结构

//中心集合
ArrayList<ArrayList<Double>> centers = new ArrayList<ArrayList<Double>>();
//all
ArrayList<ArrayList<Double>> wines = new ArrayList<ArrayList<Double>>();
//距离最近的
ArrayList<ArrayList<Double>>[] cluster;
//用k个中心
int k = 0;
//迭代次数
int count = 0;

读取wine.data文件

public k_means3(String path) {
	//read file
	try {  
        BufferedReader reader = new BufferedReader(new FileReader(new File(path)));  
        String s = null;  
        while((s = reader.readLine())!=null){  
            ArrayList<Double> temp =  textToArray(s);
            wines.add(temp);
        }  
    } catch (Exception e) {  
        // TODO Auto-generated catch block  
        e.printStackTrace();  
    } 
}

public static ArrayList<Double> textToArray(String text){
    ArrayList<Double> list = new ArrayList<Double>();
    String[] fileds = text.toString().split(",");
    for(int i=0; i<fileds.length; i++){
        list.add(Double.parseDouble(fileds[i]));
    }
    return list;
}

设置K和初始中心点

public void intialCenters() {
	int length = wines.size();  
    int dis = length/k;  
    for(int i=0; i<k; i++){  
    	//初始中心
        //wineCenter[i]=wineAll.get(dis*i); 
    	ArrayList<Double> tmp = wines.get(i*dis);
        centers.add(tmp);
    }  
    System.out.println("K是："+k+"\n初始确定的中心点是：");  
    for (int i = 0; i < k; i++) {  
        //double[] dou = centers.get(i);  
        for (int j = 0; j < 14; j++) {  
        	double x = centers.get(i).get(j);
            System.out.print(x+"   ");  
        }  
        System.out.println();  
    }  System.out.println();
}

public void setK(int k){  
    this.k = k;  
    cluster = new ArrayList[k];
    for(int i=0; i<k; i++) {
    	cluster[i] = new ArrayList<ArrayList<Double>>();
    }
    
}

计算过程

  public void work() {
  	calcu();
  	//完成输出
  	//中心点
  	for(int i=0; i<k; i++) {
  		int size6 = centers.get(0).size();
  		for(int j=1; j<size6; j++) {
  			System.out.print(centers.get(i).get(j)+" ");
  		}System.out.println();
  	}System.out.println();
  	//
  	PrintStream ps;
try {
	ps = new PrintStream("/home/huii/Desktop/kmeans2.txt");
	System.setOut(ps);
	for(int i=0; i<k; i++) {
   		int size2 = cluster[i].size();
   		int size3 = cluster[i].get(0).size();
   		for(int j=0; j<size2; j++) {
   			double t = i+1;
   			System.out.print(t+" ");
   			for(int z=0; z<size3; z++) {
   				System.out.print(cluster[i].get(j).get(z)+" ");
   			}System.out.println();
   		}
   	}
} catch (FileNotFoundException e) {
	// TODO Auto-generated catch block
	e.printStackTrace();
}

  	
  }
  
  public void calcu() {
  	boolean finish = false;
  	//距离最近的
  	for(int i=0; i<wines.size(); i++) {
  		int kind = 0;
  		double mindis = Double.MAX_VALUE;
  		for(int j=0; j<k; j++) {
  			double dis = 0;
  			int size4 = wines.get(0).size();
  			for(int z=0; z<size4; z++) {
  				double a = wines.get(i).get(z);
  				double b = centers.get(j).get(z);
  				dis += Math.pow((b-a)/(b+a), 2);
  			}
  			if(dis<mindis) {
  				mindis = dis;
  				kind = j;
  			}
  		}
  		ArrayList<Double> tmp2 = wines.get(i);
  		cluster[kind].add(tmp2);
  	}
  	//重新计算中心
  	ArrayList<ArrayList<Double>> newlist = new ArrayList<ArrayList<Double>>();
  	for(int m=0; m<k; m++) {
  		ArrayList<Double> newc = new ArrayList<Double>();
  		int size5 = cluster[m].get(0).size();
  		for(int n=0; n<size5; n++) {
  			//每列平均值
  			double sum = 0;
  			int size = cluster[m].size();
  			for(int c=0; c<size; c++) {
  				sum += cluster[m].get(c).get(n);
  			}
  			newc.add(sum/size);
  		}
  		newlist.add(newc);
  		//ArrayList<Double> oldlist = cluster[k].get(index);
  	}
  	//比较前后中心是否相同，判断迭代是否结束
  	int size1 = centers.size();
  	int fildsize = centers.get(0).size();
  	double cdis = 0;
  	for(int l=0; l<size1; l++) {
  		for(int o=0; o<fildsize; o++) {
  			double t1 = centers.get(l).get(o);
  			double t2 = newlist.get(l).get(o);
  			cdis += Math.pow((t1-t2)/(t1+t2), 2);
  		}
  	}
  	if(cdis == 0.0)
  		finish = true;
  	System.out.println(count);
  	count++;
  	if(!finish) {
  		//没完成
  		centers.clear();
  		for(int j=0; j<k; j++){
  			centers.add(newlist.get(j));
  			cluster[j].clear();
  		}
  		calcu();
  	}
  }

结果：

第一列是所属的类别，第二列往后是各行的数据。

---

实现后，发现要完全一致，大致看了一下，结果是对的，但是顺序不一样，(ó﹏ò｡)，不知道为什么MapReduce的输出顺序跟我想的不一样？？

非并行化的顺序是按照类别，类别中再按照数据集的顺序。

为了顺序输出，我在map中添加：

ArrayList<ArrayList<Double>> wines = new ArrayList<ArrayList<Double>>();
if(flag) {
	ArrayList<Double> x = new ArrayList<Double>();
    String[] tmp = value.toString().split(",");
    //System.out.print((centerIndex+1)+" ");
    double index = centerIndex+1;
    x.add(index);
    for(int i=0; i<tmp.length; i++) {
    	//System.out.print(tmp[i]+" ");
         double y = Double.parseDouble(tmp[i]);
         x.add(y);
    }//System.out.println();
    wines.add(x);
}

添加cleanup方法

protected void cleanup(Context context) throws IOException,
		InterruptedException {
	if(flag) {
		PrintStream ps = new PrintStream("/home/huii/Desktop/kmeans1.txt");//创建一个打印输出流，输出的目标是
		System.setOut(ps);//把创建的打印输出流赋给系统。即系统下次向 ps输出
		int size1 = wines.size();
		for(int i=1; i<k+1; i++) {
			for(int j=0; j<size1; j++) {
				if(wines.get(j).get(0) == i) {
					int size2 = wines.get(j).size();
					for(int m=0; m<size2; m++) {
						System.out.print(wines.get(j).get(m)+" ");
					}System.out.println();
				}
			}
		}
	}       	
}

嘻嘻嘻٩(๑>◡<๑)۶

赏

谢谢你请我吃糖果