基于Mahout的聚类实例

项目实战

基本概念

聚类分析是由若干模式组成的，通常，模式是一个度量的向量，或者是多维空间中的一个点。聚类分析以相似性为基础，在一个聚类中的模式之间比不在同一聚类中的模式之间具有更多的相似性。

需求分析

我的需求是分析全国各大高校在社科方向上的相似性，因此会以高校的若干维度为基础做聚类，最后在同一簇中的高校就定义为具有相似性。
具体由于数据限制的原因，我最后选择了15个维度的数据，如下：

P.S. 上述选择的10个学科是社科类别下最热门的10个学科，因此做为样本指代所有学科的情况。

代码实例

import java.io.BufferedReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.mahout.clustering.classify.WeightedVectorWritable;
import org.apache.mahout.clustering.kmeans.KMeansDriver;
import org.apache.mahout.clustering.kmeans.Kluster;
import org.apache.mahout.common.distance.EuclideanDistanceMeasure;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.NamedVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import tool.ClassifierHelper;
import tool.CsvTool;
import com.google.common.io.Closeables;
/**
 * 聚类测试
 * 1，两个输入路径：一个是数据的点；一个是初始集群。
        点的输入文件是SequenceFile(Key, VectorWritable)格式；
        而初始集群的输入文件格式是SequenceFiles(Text, Cluster | Canopy)
 * 2，每次迭代会产生一个输出目录"cluster-N"，输出文件格式为SequenceFile(Text, Cluster)，表示第N次迭代后产生的clusters。
 * 3，输出目录"clusteredPoints"，表示最终的集群结果，即每个集群中所包含的Points。
 * @author suwb
 * @since 2015-11-26
 */
public class Clustering {
    private final String CSV_PATH = "F:\\source\\clustering\\部属高校_聚类.csv";
    private final String POINTS_PATH = "F:\\source\\clustering\\testdata\\points";
    private final String CLUSTERS_PATH = "F:\\source\\clustering\\testdata\\clusters";
    private final String OUTPUT_PATH = "F:\\source\\clustering\\output";
    //预处理数据(比如处理成csv)
    public void prepareData(){
        Dao dao = new Dao();
        List<String[]> dataList = new ArrayList<String[]>();
        List<Object[]> unitList = dao.queryBySql("select c_name from t_unit where c_type='部属高校'");//有效数据76个，可以分10簇
        for(Object[] unitName : unitList){
            String SQL = "--";
            try {
                Object[] thisObject = dao.queryBySql(SQL).get(0);
                String[] thisString = new String[16];
                thisString[0] = unitName[0].toString();
                int k=0;//跳出两层循环的标记
                for(int i=1; i<16; i++){
                    if(i==3){//过滤异常数据
                        if(Integer.parseInt(thisObject[i-1].toString())==0){
                            k=1;
                            continue;
                        }
                    }
                    thisString[i] = thisObject[i-1].toString();
                }
                if(k==1){
                    continue;
                }
                dataList.add(thisString);
            } catch (Exception e) {
                System.out.println(unitName[0].toString());
            }
        }
        String[] header = {"高校名", "是否部属高校", "高校类型", "讲师数量", "副教授数量", "教授数量", "经济学申报数量", "管理学申报数量", "交叉学科/综合研究申报数量", 
                "教育学申报数量", "语言学申报数量", "法学申报数量", "艺术学申报数量", "马克思主义/思想政治教育申报数量", "中国文学申报数量", "历史学申报数量"};
        CsvTool.writeCsv(CSV_PATH, header, dataList);
    }
    //生成向量Vector
    public List<NamedVector> getVector() throws IOException{
        List<NamedVector> points = new ArrayList<NamedVector>();
        BufferedReader in = ClassifierHelper.open(CSV_PATH);//从输入的预测集文件读取数据
        //读取标题行，第一行变量名
        in.readLine();
        //读取下一行，数据行第一行
        String line = in.readLine();
        while (line != null) {
            //逐行向量化
            String[] univInfo = line.split(",");
            double[] fr = new double[univInfo.length-1];
            for(int i=0;i<univInfo.length-1;i++){
                fr[i] = Double.parseDouble(univInfo[i+1]);
            }
            Vector vector = new DenseVector(fr);
            NamedVector vec = new NamedVector(vector, univInfo[0]);
            points.add(vec);
            //读取下一行
            line = in.readLine();
        }
        Closeables.close(in, true);
        return points;
    }
    //生成SequenceFile格式的文件
    public void writePointsToFile(List<NamedVector> points, Path path, FileSystem fs, Configuration conf) throws IOException {
        SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf,
                path, Text.class, VectorWritable.class);
        VectorWritable vec = new VectorWritable();
        for (NamedVector point : points) {
            vec.set(point);
            writer.append(new Text(point.getName()), vec);
        }
        writer.close();
    }
    //读SequenceFile文件
    public void readSequenceFile(Path path, FileSystem fs, Configuration conf) throws IOException{
        SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
        Text key = new Text();
        VectorWritable value = new VectorWritable();
        while(reader.next(key, value)){
            System.out.println(key.toString() + " " + value.get().asFormatString());
        }
        reader.close();
    }
    //读cluster文件
    public void readCluster(Path path, FileSystem fs, Configuration conf) throws IOException{
        SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
        IntWritable key = new IntWritable();
        WeightedVectorWritable value = new WeightedVectorWritable();
        while (reader.next(key, value)) {
            System.out.println(value.toString() + " belongs to cluster " + key.toString());
        }
        reader.close();
    }
    //写入初始中心点
    public void setCenterPoints(Path path, FileSystem fs, Configuration conf, int k, List<NamedVector> vectors) throws IOException{
        SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, path, Text.class, Kluster.class);
        for (int i = 0; i < k; i++) {
            Vector vec = vectors.get(i);
            Kluster cluster = new Kluster(vec, i, new EuclideanDistanceMeasure());
            writer.append(new Text(cluster.getIdentifier()), cluster);
        }
        writer.close();
    }
    public void work(Configuration conf) throws IOException, ClassNotFoundException, InterruptedException{
        Path inputPath = new Path(POINTS_PATH);
        Path clustersPath = new Path(CLUSTERS_PATH);
        Path outputPath = new Path(OUTPUT_PATH);
//      HadoopUtil.delete(conf, outputPath);
//      FuzzyKMeansDriver.run(conf, inputPath, clustersPath, outputPath, new TanimotoDistanceMeasure(), 0.001, 20, 2.0f, true, true, 0.0, true);
        KMeansDriver.run(conf, inputPath, clustersPath, outputPath, new EuclideanDistanceMeasure(), 0.001, 20, true, 0.0, true);
        System.out.println("Clusters are ready");
    }
    //执行聚类程序
    public void run() throws IOException, InterruptedException, ClassNotFoundException{
//      prepareData();
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
//      writePointsToFile(getVector(), new Path(POINTS_PATH + "\\file1"), fs, conf);
        setCenterPoints(new Path(CLUSTERS_PATH + "\\part-00000"), fs, conf, 10, getVector());
        work(conf);
        readCluster(new Path(OUTPUT_PATH + "\\" + Kluster.CLUSTERED_POINTS_DIR + "\\part-m-0"), fs, conf);
    }
}

package tool;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import com.csvreader.CsvReader;
import com.csvreader.CsvWriter;
/**
 * CSV工具包
 * @author suwb
 */
public class CsvTool {
    /**
     * 写CSV文件
     * @param outFilePath   数据文件路径
     * @param header        内容第一行标题
     * @param dataList      内容
     */
    public static void writeCsv(String outFilePath, String[] header, List<String[]> dataList) {
        CsvWriter writer = null;  
        try {
            writer = new CsvWriter(outFilePath, ',', Charset.forName("UTF-8"));
            //写文件头
            if(header != null){
                writer.writeRecord(header);
            }
            //写文件内容
            for (String[] datas : dataList) {
                writer.writeRecord(datas);
            }
        } catch (IOException e) {  
            e.printStackTrace();  
        } finally {  
            writer.close();  
        }  
    }
    /** 
     * 读CSV文件
     * @param csvFilePath 
     * @throws Exception 
     */  
    public static List<Object[]> readCsv(String csvFilePath) throws Exception {  
        // 返回结果
        List<Object[]> datas = new ArrayList<Object[]>();  
        CsvReader reader = new CsvReader(csvFilePath, ',', Charset.forName("UTF-8"));
        //读文件内容
        while (reader.readRecord()) {  
            datas.add(reader.getValues());  
        }  
        return datas;
    }  
}

代码解释

首先，Mahout版本为0.8，目前最权威的参考书《Mahout in action》里是基于0.5的版本的介绍，这会有一些区别，比如，0.8中已经没有Cluster类了，替代为了Kluster；同时，0.8中没有了书中描述得基于内存（in-memory）形式的聚类算法。再则，如果你本机并没有hadoop的环境，那么在执行KMeansDriver.run或FuzzyKMeansDriver.run的时候，最后一个参数必须置为true，否则会执行不通过。其他的大致与书中介绍的一致。
聚类的步骤总结如下：
预处理数据-->生成Vector格式-->写入SenquenceFile--> 设定中心点-->执行聚类程序-->结果分析

最后

最后得到的典型结果如下：

簇0有：合肥工业大学、重庆大学、河海大学、中国矿业大学、同济大学、华东理工大学、西安交通大学、大连理工大学、天津大学、中国海洋大学
簇1有：北京科技大学、西安电子科技大学、北京化工大学、中国地质大学（北京）、北京邮电大学、中国石油大学（北京）、中国矿业大学（北京）、中国石油大学（华东）、华北电力大学
簇2有：上海交通大学、中南大学、武汉理工大学、兰州大学、东南大学、四川大学、湖南大学、江南大学、华中科技大学、浙江大学、华南理工大学
簇3有：中央戏剧学院、中央美术学院、中央音乐学院、国际关系学院
簇4有：南京农业大学、北京林业大学、西北农林科技大学、华中农业大学、东北林业大学、中国农业大学
簇5有：上海财经大学、中央财经大学、西南财经大学、对外经济贸易大学、陕西师范大学、中南财经政法大学、中国政法大学
簇6有：北京外国语大学、北京语言大学、中国药科大学、北京中医药大学、中国传媒大学、上海外国语大学
簇7有：东华大学、长安大学、东北大学、北京交通大学、西南交通大学、中国地质大学（武汉）、电子科技大学、清华大学
簇8有：武汉大学、厦门大学、中国人民大学、南京大学、南开大学、山东大学、北京大学、复旦大学、吉林大学、中山大学、西南大学
簇9有：华东师范大学、华中师范大学、东北师范大学、北京师范大学

从结果来看，这个聚类准确度还是非常高的，由此我们也可以知道，在社科方面，我校与很多知名偏理工类学校都很一般，即便是清华也只是跟电子科大等大学比肩，而武大却是跟北大统一梯队，站稳了国内社科类的第一把交椅。
在写这个例子的时候，还是碰到了很多困难的，因为我主要就是在参考《Mahout in action》，而0.5到0.8的演变中还是有了相当的变化，所以经常会碰到一些异常，这个时候就必须得发挥查找资料的能力了，在完成这个例子的过程中，我主要得到了以下几个帮助，希望可以让其他人避免走弯路。
1、http://qnalist.com/questions/4878396/in-memory-kmeans-clustering
2、http://stackoverflow.com/questions/9565998/getting-an-ioexception-when-running-a-sample-code-in-mahout-in-action-on-mahou
3、http://ghost-face.iteye.com/blog/1905255
4、http://www.tuicool.com/articles/ryiEN3