mahout算法canopy源码分析之一：获得输入数据-其它-@大数据资讯

对于canopy的输入数据需要的形式为序列文件，同时保证key：Text、value：VectorWritable。昨晚准备打算使用单纯的java程序搞定输入数据的准备，无奈老是会出点问题，昨晚的问题“找不到文件”暂时还没找到原因。

其实如果只是要获得输入数据那么，可以使用mahout官网提供的方法在得到了序列的*.txt文件后直接把mahout-distribution-0.7.zip解压拷贝到虚拟机，（在/etc/profile里面配置下hadoop_home变量）然后找到mahout_home/bin目录，执行 chmod +x mahout ，然后分别执行

[python]view plaincopy 
./mahout seqdirectory -i <input> -o <output>  

[python]view plaincopy 
./mahout seq2sparse -i <output>/chunk-0 -o <output-in>  

上面的<input>、<output>对应于自己的输入和输出；我所使用的数据并不是 Reuters dataset的全部数据，而只是前三个：reut2-000.sgm、reut2-001.sgm、reut2-002.sgm，这样的数据在经过ExtractReuters 后生成了3000个文件，然后经过seqdirectory合并成了一个2.41M的数据文件。seq2sparse有7个job，每个job负责自己的内容，这个暂时不加分析；最终的结果在<output-in>/tfidf-vectors里面，即为输入数据；

有了输入数据就可以直接跑程序了，首先不管程序是什么样子的，先跑出结果再说：

[java]view plaincopy 
package mahout.test.canopy;  
import java.io.IOException;  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.Path;  
import org.apache.mahout.clustering.canopy.CanopyDriver;  
import org.apache.mahout.common.distance.DistanceMeasure;  
import org.apache.mahout.common.distance.EuclideanDistanceMeasure;  
public class CanopyTest {  
    public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {  
        Configuration conf =new Configuration();  
        conf.set("mapred.job.tracker", "192.168.128.138:9001");  
        Path input=new Path("hdfs://hadoop:9000/user/hadoop/output/canopyvec/tfidf-vectors");  
        Path output=new Path("hdfs://hadoop:9000/user/hadoop/output/canopy-output");  
        DistanceMeasure measure=new EuclideanDistanceMeasure();   
        CanopyDriver.buildClusters(conf, input, output, measure, 33.1, 22.1, 3, false);  
        System.out.println("job is done.");  
    }  
}  

最开始的时候我的t1、t2设置为3.1、2.1结果map出来的结果数是为零（这个暂时也不知道是代表什么意思），后来改为上面的结果可以看到map输出了509条记录，reduece输出3条记录（符合参数clusterFileter的设定值：3），最后的输出为：canopy-output/clusters-0-final/part-r-00000。

为了便于后面的观察，所以不使用上面的数据，而使用自己造的数据，造数据之前首先要知道输入数据的格式，那么使用下面的代码看下输入数据是什么：

[java]view plaincopy 
package mahout.test.utils;  
import java.io.IOException;  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Job;  
import org.apache.hadoop.mapreduce.Mapper;  
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
import org.apache.hadoop.util.ToolRunner;  
import org.apache.mahout.common.AbstractJob;  
import org.apache.mahout.math.Vector;  
import org.apache.mahout.math.VectorWritable;  
/** 
 * 读取序列文件，输入Key：Text，Value：VectorWritable 
 * @author fansy 
 * @version 2013/7/21 11:32 
 * 
 */  
public class ReadTextVectorWritable extends AbstractJob {  
    @Override  
    public int run(String[] arg0) throws Exception {  
        if(arg0.length!=6||(!"-i".equals(arg0[0])||(!"-o".equals(arg0[2]))||(!"-jt".equals(arg0[4])))){  
            System.err.println("参数不正确！");  
            System.out.println("Usage:");  
            System.out.println("-i <input> -o <output> -jt <jobtracker:port>");  
            System.exit(-1);  
        }         
        Configuration conf=new Configuration();  
        conf.set("mapred.job.tracker", arg0[5]);  
        Job job=new Job(conf);  
        job.setJobName("readTextVector with Input:"+arg0[1]);  
        job.setJarByClass(ReadTextVectorWritable.class);  
        job.setInputFormatClass(SequenceFileInputFormat.class);       
        job.setMapperClass(RM.class);  
        job.setMapOutputKeyClass(Text.class);  
        job.setMapOutputValueClass(Text.class);  
        job.setNumReduceTasks(0);  
        SequenceFileInputFormat.addInputPath(job, new Path(arg0[1]));  
        FileOutputFormat.setOutputPath(job, new Path(arg0[3]));       
        return job.waitForCompletion(true)? 0:-1;  
    }     
    public static class RM extends Mapper<Text,VectorWritable,Text,Text>{  
        public void map(Text key,VectorWritable value,Context context)throws InterruptedException,IOException{  
            Vector vector=value.get();  
            context.write(key, new Text(vector.asFormatString()));  
        }  
    }  
    public static void main(String[] args) throws Exception{  
        ToolRunner.run(new Configuration(), new ReadTextVectorWritable(), args);  
    }  
}  

然后查看hdfs上面的输入数据，如下：

对于上面的数据，其实应该和下面的数据效果是一样的：

[html]view plaincopy 
1    {1:3.45，2:4.67,3:2.34}  
2    {1:4.65，3:4.62,3:4.34}  
3    {1:5.95，5:4.67,3:2.24}  

第一个代表一个样本号，冒号前面代表维度，冒号后面代表相应维度的大小。猜测：对于reuters数据，应该是把所有的单词全部排序，然后按照一定的规则进行编码（可以是从零到n的编码，所以冒号前面的数字就代表某一个单词），后面的数值应该是该单词在文件中的重要性（？这个应该是有一个pagerank的算法之类的）；

有了上面的想法后就可以直接构造一定的输入数据（自己构造少量数据就可以对算法有很清晰的认识），然后把canopy算法的代码改为java代码（不使用hadoop就可以跑的代码），然后就可以进行调试分析其算法逻辑了。