import org.apache.mahout.common.Pair; //导入依赖的package包/类
public double loadResult(Path outputDirPath, Configuration conf) throws IOException {
  Path finalNumberFile = new Path(outputDirPath, "part-r-00000");
  SequenceFileIterator<NullWritable, DoubleWritable> iterator = 
      new SequenceFileIterator<NullWritable, DoubleWritable>(
      finalNumberFile, true, conf);
  double norm2;
  try {
    Pair<NullWritable, DoubleWritable> next = iterator.next();
    norm2 = next.getSecond().get();
    if (iterator.hasNext())
      throw new IOException("More than one value after norm2Job!");
  } finally {
    Closeables.close(iterator, false);
  }
  return norm2;
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
private static String[] loadDictionary(String dictionaryPath, Configuration conf) {
  if (dictionaryPath == null) {
    return null;
  }
  Path dictionaryFile = new Path(dictionaryPath);
  List<Pair<Integer, String>> termList = Lists.newArrayList();
  int maxTermId = 0;
   // key is word value is id
  for (Pair<Writable, IntWritable> record
          : new SequenceFileIterable<Writable, IntWritable>(dictionaryFile, true, conf)) {
    termList.add(new Pair<Integer, String>(record.getSecond().get(),
        record.getFirst().toString()));
    maxTermId = Math.max(maxTermId, record.getSecond().get());
  }
  String[] terms = new String[maxTermId + 1];
  for (Pair<Integer, String> pair : termList) {
    terms[pair.getFirst()] = pair.getSecond();
  }
  return terms;
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
/**
 * @param topicModelStateTemp
 * @param iteration
 * @return {@code double[2]} where first value is perplexity and second is model weight of those
 *         documents sampled during perplexity computation, or {@code null} if no perplexity data
 *         exists for the given iteration.
 * @throws IOException
 */
public static double readPerplexity(Configuration conf, Path topicModelStateTemp, int iteration)
  throws IOException {
  Path perplexityPath = perplexityPath(topicModelStateTemp, iteration);
  FileSystem fs = FileSystem.get(perplexityPath.toUri(), conf);
  if (!fs.exists(perplexityPath)) {
    log.warn("Perplexity path {} does not exist, returning NaN", perplexityPath);
    return Double.NaN;
  }
  double perplexity = 0;
  double modelWeight = 0;
  long n = 0;
  for (Pair<DoubleWritable, DoubleWritable> pair : new SequenceFileDirIterable<DoubleWritable, DoubleWritable>(
      perplexityPath, PathType.LIST, PathFilters.partFilter(), null, true, conf)) {
    modelWeight += pair.getFirst().get();
    perplexity += pair.getSecond().get();
    n++;
  }
  log.info("Read {} entries with total perplexity {} and model weight {}", new Object[] { n,
          perplexity, modelWeight });
  return perplexity / modelWeight;
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
public static void pruneVectors(Path tfDir, Path prunedTFDir, Path prunedPartialTFDir, long maxDF,
                                Configuration baseConf,
                                Pair<Long[], List<Path>> docFrequenciesFeatures,
                                float normPower,
                                boolean logNormalize,
                                int numReducers) throws IOException, InterruptedException, ClassNotFoundException {

  int partialVectorIndex = 0;
  List<Path> partialVectorPaths = new ArrayList<Path>();
  for (Path path : docFrequenciesFeatures.getSecond()) {
    Path partialVectorOutputPath = new Path(prunedPartialTFDir, "partial-" + partialVectorIndex++);
    partialVectorPaths.add(partialVectorOutputPath);
    pruneVectorsPartial(tfDir, partialVectorOutputPath, path, maxDF, baseConf);
  }

  mergePartialVectors(partialVectorPaths, prunedTFDir, baseConf, normPower, logNormalize, numReducers);
  HadoopUtil.delete(new Configuration(baseConf), prunedPartialTFDir);
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Override
protected void setup(Context context) throws IOException, InterruptedException {
  super.setup(context);
  Configuration conf = context.getConfiguration();
  URI[] localFiles = DistributedCache.getCacheFiles(conf);
  Preconditions.checkArgument(localFiles != null && localFiles.length >= 1,
          "missing paths from the DistributedCache");

  dimension = conf.getInt(PartialVectorMerger.DIMENSION, Integer.MAX_VALUE);
  sequentialAccess = conf.getBoolean(PartialVectorMerger.SEQUENTIAL_ACCESS, false);
  namedVector = conf.getBoolean(PartialVectorMerger.NAMED_VECTOR, false);
  maxNGramSize = conf.getInt(DictionaryVectorizer.MAX_NGRAMS, maxNGramSize);

  Path dictionaryFile = new Path(localFiles[0].getPath());
  // key is word value is id
  for (Pair<Writable, IntWritable> record
          : new SequenceFileIterable<Writable, IntWritable>(dictionaryFile, true, conf)) {
    dictionary.put(record.getFirst().toString(), record.getSecond().get());
  }
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
/**
 * Calculates the document frequencies of all terms from the input set of vectors in
 * {@link SequenceFile} format. This job uses a fixed limit on the maximum memory used by the feature chunk
 * per node thereby splitting the process across multiple map/reduces.
 * 
 * @param input
 *          input directory of the vectors in {@link SequenceFile} format
 * @param output
 *          output directory where document frequencies will be stored
 * @param chunkSizeInMegabytes
 *          the size in MB of the feature => id chunk to be kept in memory at each node during Map/Reduce
 *          stage. Its recommended you calculated this based on the number of cores and the free memory
 *          available to you per node. Say, you have 2 cores and around 1GB extra memory to spare we
 *          recommend you use a split size of around 400-500MB so that two simultaneous reducers can create
 *          partial vectors without thrashing the system due to increased swapping
 */
public static Pair<Long[],List<Path>> calculateDF(Path input,
                                                  Path output,
                                                  Configuration baseConf,
                                                  int chunkSizeInMegabytes)
  throws IOException, InterruptedException, ClassNotFoundException {

  if (chunkSizeInMegabytes < MIN_CHUNKSIZE) {
    chunkSizeInMegabytes = MIN_CHUNKSIZE;
  } else if (chunkSizeInMegabytes > MAX_CHUNKSIZE) { // 10GB
    chunkSizeInMegabytes = MAX_CHUNKSIZE;
  }

  Path wordCountPath = new Path(output, WORDCOUNT_OUTPUT_FOLDER);

  startDFCounting(input, wordCountPath, baseConf);

  return createDictionaryChunks(wordCountPath, output, baseConf, chunkSizeInMegabytes);
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Override
protected void setup(Context context) throws IOException, InterruptedException {
  super.setup(context);
  Configuration conf = context.getConfiguration();
  URI[] localFiles = DistributedCache.getCacheFiles(conf);
  Preconditions.checkArgument(localFiles != null && localFiles.length >= 1, 
      "missing paths from the DistributedCache");

  vectorCount = conf.getLong(TFIDFConverter.VECTOR_COUNT, 1);
  featureCount = conf.getLong(TFIDFConverter.FEATURE_COUNT, 1);
  minDf = conf.getInt(TFIDFConverter.MIN_DF, 1);
  maxDf = conf.getLong(TFIDFConverter.MAX_DF, -1);
  sequentialAccess = conf.getBoolean(PartialVectorMerger.SEQUENTIAL_ACCESS, false);
  namedVector = conf.getBoolean(PartialVectorMerger.NAMED_VECTOR, false);

  Path dictionaryFile = new Path(localFiles[0].getPath());
  // key is feature, value is the document frequency
  for (Pair<IntWritable,LongWritable> record 
       : new SequenceFileIterable<IntWritable,LongWritable>(dictionaryFile, true, conf)) {
    dictionary.put(record.getFirst().get(), record.getSecond().get());
  }
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Override
protected void setup(Context context) throws IOException, InterruptedException {
  super.setup(context);
  Configuration conf = context.getConfiguration();
  URI[] localFiles = DistributedCache.getCacheFiles(conf);
  Preconditions.checkArgument(localFiles != null && localFiles.length >= 1,
          "missing paths from the DistributedCache");

  maxDf = conf.getLong(HighDFWordsPruner.MAX_DF, -1);

  Path dictionaryFile = new Path(localFiles[0].getPath());
  // key is feature, value is the document frequency
  for (Pair<IntWritable, LongWritable> record :
          new SequenceFileIterable<IntWritable, LongWritable>(dictionaryFile, true, conf)) {
    dictionary.put(record.getFirst().get(), record.getSecond().get());
  }
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
protected RuleBase processOutput(JobContext job, Path outputPath) throws IOException {
 
  Configuration conf = job.getConfiguration();

  FileSystem fs = outputPath.getFileSystem(conf);

  Path[] outfiles = Chi_RWCSUtils.listOutputFiles(fs, outputPath);
  
  RuleBase ruleBase = null;
  
  // read all the outputs
  for (Path path : outfiles) {
    for (Pair<LongWritable,RuleBase> record : new SequenceFileIterable<LongWritable, RuleBase>(path, conf)) {
  	if(ruleBase == null){
        ruleBase = record.getSecond();
  	}
    }
  }
  
  return ruleBase;
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Override
protected Pair<K,V> computeNext() {
  if (!reuseKeyValueInstances || value == null) {
    key = ReflectionUtils.newInstance(keyClass, conf);
    if (!noValue) {
      value = ReflectionUtils.newInstance(valueClass, conf);
    }
  }
  try {
    boolean available;
    if (noValue) {
      available = reader.next(key);
    } else {
      available = reader.next(key, value);
    }
    if (!available) {
      close();
      return null;
    }
    return new Pair<K,V>(key, value);
  } catch (IOException ioe) {
    close();
    throw new IllegalStateException(ioe);
  }
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
public StableFixedSizeSamplingIterator(int size, Iterator<T> source) {
  List<Pair<Integer,T>> buf = Lists.newArrayListWithCapacity(size);
  int sofar = 0;
  Random random = RandomUtils.getRandom();
  while (source.hasNext()) {
    T v = source.next();
    sofar++;
    if (buf.size() < size) {
      buf.add(new Pair<Integer,T>(sofar, v));
    } else {
      int position = random.nextInt(sofar);
      if (position < buf.size()) {
        buf.set(position, new Pair<Integer,T>(sofar, v));
      }
    }
  }

  Collections.sort(buf);
  delegate = Iterators.transform(buf.iterator(),
                                 new Function<Pair<Integer,T>,T>() {
                                   @Override
                                   public T apply(Pair<Integer,T> from) {
                                     return from.getSecond();
                                   }
                                 });
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
public static void move(MyVectorInput input, MyVectorOutput output) throws IOException{
    
    //move the dictionary
    Pair<String,Integer> dictEntry;
    int terms=0, vectors=0;
    
    while((dictEntry=input.nextDictEntry())!=null){
        output.writeDictEntry(dictEntry);
        terms++;
    }
    
    System.out.println("Moved "+terms+" dict terms");
    
    //move the vectors
    MySparseVector vector;
    while((vector=input.nextVector())!=null){
        output.writeVector(vector);
        vectors++;
    }
    
    output.close();
    
    System.out.println("Moved "+ vectors+ " document vectors");
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
private static Pair<Integer,Double> popMin(MySparseVector v){
    int minIndexIndex=-1, minIndexValue=Integer.MAX_VALUE;
    double minValue=-1;
    
    for(int i=0; i<v.size(); i++){
        if(minIndexValue>v.indices.get(i)){
            minIndexIndex=i;
            minIndexValue=v.indices.get(i);
            minValue=v.values.get(i);
        }
    }
    
    v.indices.remove(minIndexIndex);
    v.values.remove(minIndexIndex);
    return new Pair(minIndexValue, minValue);
    
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Override
public Pair<String, Integer> nextDictEntry() throws IOException {
    String line, term;
    while((line=reader.readLine())!=null){
        if(!line.startsWith("@attribute"))
           if(line.startsWith("@data")) break; //end of dictinary
           else continue;
        if(line.contains("@@[email protected]@")) continue;
        
        //handle the new term
        termCount++; //new term
        
        //strip the term
        int beginningOfTerm=11; // "@attribute" is 10 chars long, we will strip this and the following whitespace
        int endOfTerm=line.lastIndexOf("numeric")-1;
        term = line.substring(beginningOfTerm, endOfTerm);
        //return the new term with its id
        return new Pair(term, termCount);            
        
    }
    return null;
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Override
public MySparseVector nextVector() throws IOException {
   
    if(!vecIterator.hasNext()) return null;
    
    Pair<Text, VectorWritable> entry = vecIterator.next();
    
    String name = entry.getFirst().toString();
    VectorWritable mahoutVector = entry.getSecond();
    
    ArrayList<Integer> indices = new ArrayList();
    ArrayList<Double> values = new ArrayList();
    
    for(Element e: mahoutVector.get().all()){
        double value =e.get();
        if (value==0) continue;
        values.add(value);
        int index= e.index();
        indices.add(index);            
    }
    
    return new MySparseVector(indices, values);
    
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
@Test
public void testPathField() throws Exception {
  Configuration conf = mapDriver.getConfiguration();
  conf.set(RegexIngestMapper.REGEX, "\\w+");
  conf.set(RegexIngestMapper.GROUPS_TO_FIELDS, "0=body");

  FileSystem fs = FileSystem.get(conf);
  Path dir = new Path(fs.getWorkingDirectory(), "build");
  Path sub = new Path(dir, "REGIMT");
  Path tempDir = new Path(sub, "tmp-dir");
  fs.mkdirs(tempDir);

  // First File
  String splitFilePath = "/path/to/file1.txt";
  mapDriver.setMapInputPath(new Path(splitFilePath));
  mapDriver.withInput(new Text("0"), new Text("File content one"));
  List<org.apache.hadoop.mrunit.types.Pair<Text, LWDocumentWritable>> run1 = mapDriver.run();
  assertTrue(run1.size() == 1);
  LWDocument doc = run1.get(0).getSecond().getLWDocument();
  assertNotNull("document is null", doc);
  // TODO: Check Fields

}
 

import org.apache.mahout.common.Pair; //导入依赖的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.common.Pair; //导入依赖的package包/类
protected RuleBase processOutput(JobContext job, Path outputPath) throws IOException {
  	  
  Configuration conf = job.getConfiguration();

  FileSystem fs = outputPath.getFileSystem(conf);

  Path[] outfiles = Chi_RWUtils.listOutputFiles(fs, outputPath);
  
  RuleBase ruleBase = null;
  
  // read all the outputs
  for (Path path : outfiles) {
    for (Pair<LongWritable,RuleBase> record : new SequenceFileIterable<LongWritable, RuleBase>(path, conf)) {
  	if(ruleBase == null){
        ruleBase = record.getSecond();
  	}
    }
  }
  
  return ruleBase;
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
/**
 * After running the job, we can load the results from HDFS with this method
 * 
 * @param meanSpanPath
 *          the path to the single file having the results
 * @param normalizeMean
 *          normalize the mean as well
 * @param conf
 *          the configuration
 * @throws IOException
 *           when face problem parsing the result file
 */
public void loadResults(Path meanSpanPath, boolean normalizeMean, Configuration conf)
    throws IOException {
  SequenceFileIterator<IntWritable, VectorWritable> iterator = 
      new SequenceFileIterator<IntWritable, VectorWritable>(
      meanSpanPath, true, conf);
  try {
    Pair<IntWritable, VectorWritable> next;
    next = iterator.next();
    if (next.getFirst().get() == MEANVECTOR)
      meanVector = next.getSecond().get();
    else
      spanVector = next.getSecond().get();
    next = iterator.next();
    if (next.getFirst().get() == MEANVECTOR)
      meanVector = next.getSecond().get();
    else
      spanVector = next.getSecond().get();
  } finally {
    Closeables.close(iterator, false);
  }
  if (normalizeMean)
    meanVector.assign(spanVector, new DoubleDoubleFunction() {
      @Override
      public double apply(double v, double span) {
        return v / (span != 0 ? span : 1);
      }
    });
}
 

import org.apache.mahout.common.Pair; //导入依赖的package包/类
public void loadResult(Path outDirPath, Configuration conf)
    throws IOException {
  Path finalNumberFile = new Path(outDirPath, "part-r-00000");
  SequenceFileIterator<NullWritable, DoubleWritable> iterator = new SequenceFileIterator<NullWritable, DoubleWritable>(
      finalNumberFile, true, conf);
  try {
    Pair<NullWritable, DoubleWritable> next = iterator.next();
    finalSum = next.getSecond().get();
  } finally {
    Closeables.close(iterator, false);
  }
}