import org.apache.mahout.clustering.classify.WeightedVectorWritable; //导入依赖的package包/类
/**
 * This method takes the clustered points output by the clustering algorithms as input and writes them into
 * their respective clusters.
 */
public void process() throws IOException {
  createPostProcessDirectory();
  for (Pair<?,WeightedVectorWritable> record : 
       new SequenceFileDirIterable<Writable,WeightedVectorWritable>(clusteredPoints,
                                                                    PathType.GLOB,
                                                                    PathFilters.partFilter(),
                                                                    null,
                                                                    false,
                                                                    conf)) {
    String clusterId = record.getFirst().toString().trim();
    putVectorInRespectiveCluster(clusterId, record.getSecond());
  }
  IOUtils.close(writersForClusters.values());
  writersForClusters.clear();
}
 

import org.apache.mahout.clustering.classify.WeightedVectorWritable; //导入依赖的package包/类
/**
 * Cluster the data using Hadoop
 */
private static void clusterDataMR(Path input, Path clustersIn, Path output)
    throws IOException, InterruptedException, ClassNotFoundException {
  Configuration conf = new Configuration();
  conf.set(STATE_IN_KEY, clustersIn.toString());
  Job job = new Job(conf,
      "Mean Shift Driver running clusterData over input: " + input);
  job.setOutputKeyClass(IntWritable.class);
  job.setOutputValueClass(WeightedVectorWritable.class);
  job.setMapperClass(MeanShiftCanopyClusterMapper.class);

  job.setInputFormatClass(SequenceFileInputFormat.class);
  job.setOutputFormatClass(SequenceFileOutputFormat.class);
  job.setNumReduceTasks(0);
  job.setJarByClass(MeanShiftCanopyDriver.class);

  FileInputFormat.setInputPaths(job, input);
  FileOutputFormat.setOutputPath(job, output);

  if (!job.waitForCompletion(true)) {
    throw new InterruptedException(
        "Mean Shift Clustering failed on clustersIn " + clustersIn);
  }
}
 

import org.apache.mahout.clustering.classify.WeightedVectorWritable; //导入依赖的package包/类
/**
 * 
 * Finds out the cluster directory of the vector and writes it into the specified cluster.
 */
private void putVectorInRespectiveCluster(String clusterId, WeightedVectorWritable point) throws IOException {
  Writer writer = findWriterForVector(clusterId);
  postProcessedClusterDirectories.put(clusterId,
                                      PathDirectory.getClusterPathForClusterId(clusterPostProcessorOutput, clusterId));
  writeVectorToCluster(writer, point);
}
 

import org.apache.mahout.clustering.classify.WeightedVectorWritable; //导入依赖的package包/类
@Override
 protected void map(WritableComparable<?> key, ClusterWritable clusterWritable, Context context)
   throws IOException, InterruptedException {
   // canopies use canopyIds assigned when input vectors are processed as vectorIds too
MeanShiftCanopy canopy = (MeanShiftCanopy)clusterWritable.getValue();
   int vectorId = canopy.getId();
   for (MeanShiftCanopy msc : canopies) {
     for (int containedId : msc.getBoundPoints().toList()) {
       if (vectorId == containedId) {
         context.write(new IntWritable(msc.getId()),
                        new WeightedVectorWritable(1, canopy.getCenter()));
       }
     }
   }
 }
 

import org.apache.mahout.clustering.classify.WeightedVectorWritable; //导入依赖的package包/类
/**
 * The key is the cluster id and the value is the vector.
 */
@Override
protected void map(IntWritable key, WeightedVectorWritable vector, Context context) throws IOException,
                                                                                   InterruptedException {
  context.write(new Text(key.toString().trim()), new VectorWritable(vector.getVector()));
}
 

import org.apache.mahout.clustering.classify.WeightedVectorWritable; //导入依赖的package包/类
/**
 * Writes vector to the cluster directory.
 */
private void writeVectorToCluster(Writer writer, WeightedVectorWritable point) throws IOException {
  writer.append(new LongWritable(uniqueVectorId++), new VectorWritable(point.getVector()));
  writer.sync();
}