import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
public void reduce(CompositeWritable compositeId,
    Iterable<VectorWritable> vectors, Context context) throws IOException,
    InterruptedException {
  Iterator<VectorWritable> it = vectors.iterator();
  if (!it.hasNext()) {
    return;
  }
  //All XtX rows are mapped to the same key
  if (compositeId.isXtX()) {
    writeXtXToFile(vectors, xtxOutputPath);
    return;
  }
  //Reduce YtX
  Vector accumulator = it.next().get();
  while (it.hasNext()) {
    Vector row = it.next().get();
    accumulator.assign(row, Functions.PLUS);
  }
  iw.set(compositeId.rowId);
  vw.set(accumulator);
  context.write(iw, vw);
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
public void observe(Vector x, double weight) {
  s0 += weight;
  Vector weightedX = x.times(weight);
  if (s1 == null) {
    s1 = weightedX;
  } else {
    s1.assign(weightedX, Functions.PLUS);
  }
  Vector x2 = x.times(x).times(weight);
  if (s2 == null) {
    s2 = x2;
  } else {
    s2.assign(x2, Functions.PLUS);
  }
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
public void observe(Vector x, double weight) {
  if (weight == 1.0) {
    observe(x);
  } else {
    setS0(getS0() + weight);
    Vector weightedX = x.times(weight);
    if (getS1() == null) {
      setS1(weightedX);
    } else {
      getS1().assign(weightedX, Functions.PLUS);
    }
    Vector x2 = x.times(x).times(weight);
    if (getS2() == null) {
      setS2(x2);
    } else {
      getS2().assign(x2, Functions.PLUS);
    }
  }
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
/**
 * Helper method, performs the actual calculation. Looks something like this:
 *
 * (log(2) / lambda_k * log(lambda_k) * log(lambda_k^beta0 / 2)) * [
 * - (((u_i / sqrt(d_i)) - (u_j / sqrt(d_j)))^2 + (1 - lambda) * 
 *   ((u_i^2 / d_i) + (u_j^2 / d_j))) ]
 */
private double performSensitivityCalculation(double eigenvalue,
                                             double evi,
                                             double evj,
                                             double diagi,
                                             double diagj) {
  
  double firsthalf = Functions.LOGARITHM.apply(2)
      / (eigenvalue * Functions.LOGARITHM.apply(eigenvalue)
         * Functions.LOGARITHM.apply(Functions.POW.apply(eigenvalue, beta0) / 2));
  
  double secondhalf =
      -Functions.POW.apply(evi / Functions.SQRT.apply(diagi) - evj / Functions.SQRT.apply(diagj), 2)
      + (1.0 - eigenvalue) * (Functions.POW.apply(evi, 2) / diagi + Functions.POW.apply(evj, 2) / diagj);
  
  return firsthalf * secondhalf;
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
public TopicModel(Matrix topicTermCounts, Vector topicSums, double eta, double alpha,
  String[] dictionary, int numThreads, double modelWeight) {
  this.dictionary = dictionary;
  this.topicTermCounts = topicTermCounts;
  this.topicSums = topicSums;
  this.numTopics = topicSums.size();
  this.numTerms = topicTermCounts.numCols();
  this.eta = eta;
  this.alpha = alpha;
  this.sampler = new Sampler(RandomUtils.getRandom());
  this.numThreads = numThreads;
  if (modelWeight != 1) {
    topicSums.assign(Functions.mult(modelWeight));
    for (int x = 0; x < numTopics; x++) {
      topicTermCounts.viewRow(x).assign(Functions.mult(modelWeight));
    }
  }
  initializeThreadPool();
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
public void trainDocTopicModel(Vector original, Vector topics, Matrix docTopicModel) {
  // first calculate p(topic|term,document) for all terms in original, and all topics,
  // using p(term|topic) and p(topic|doc)
  pTopicGivenTerm(original, topics, docTopicModel);
  normalizeByTopic(docTopicModel);
  // now multiply, term-by-term, by the document, to get the weighted distribution of
  // term-topic pairs from this document.
  Iterator<Vector.Element> it = original.iterateNonZero();
  while (it.hasNext()) {
    Vector.Element e = it.next();
    for (int x = 0; x < numTopics; x++) {
      Vector docTopicModelRow = docTopicModel.viewRow(x);
      docTopicModelRow.setQuick(e.index(), docTopicModelRow.getQuick(e.index()) * e.get());
    }
  }
  // now recalculate p(topic|doc) by summing contributions from all of pTopicGivenTerm
  topics.assign(0.0);
  for (int x = 0; x < numTopics; x++) {
    topics.set(x, docTopicModel.viewRow(x).norm(1));
  }
  // now renormalize so that sum_x(p(x|doc)) = 1
  topics.assign(Functions.mult(1/topics.norm(1)));
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
protected void reduce(WritableComparable<?> key, Iterable<VectorWritable> values, Context context) throws IOException,
        InterruptedException {

  Vector vector = null;
  for (VectorWritable value : values) {
    if (vector == null) {
      vector = value.get().clone();
      continue;
    }
    //value.get().addTo(vector);
    vector.assign(value.get(), Functions.PLUS);
  }

  if (normPower != PartialVectorMerger.NO_NORMALIZING) {
    if (logNormalize) {
      vector = vector.logNormalize(normPower);
    } else {
      vector = vector.normalize(normPower);
    }
  }

  VectorWritable vectorWritable = new VectorWritable(vector);
  context.write(key, vectorWritable);
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
/**
 * The mapper computes a running sum of the vectors the task has seen.
 * Element 0 of the running sum vector contains a count of the number of
 * vectors that have been seen. The remaining elements contain the
 * column-wise running sum. Nothing is written at this stage
 */
@Override
public void map(Writable r, VectorWritable v, Context context)
		throws IOException {
	if (runningSum == null) {
		/*
		 * If this is the first vector the mapper has seen, instantiate
		 * a new vector using the parameter VECTOR_CLASS
		 */
		runningSum = ClassUtils.instantiateAs(vectorClass,
				Vector.class, new Class<?>[] { int.class },
				new Object[] { v.get().size() + 1 });
		runningSum.set(0, 1);
		runningSum.viewPart(1, v.get().size()).assign(v.get());
	} else {
		runningSum.set(0, runningSum.get(0) + 1);
		runningSum.viewPart(1, v.get().size()).assign(v.get(),
				Functions.PLUS);
	}
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
public void reduce(IntWritable rowNum, Iterator<VectorWritable> it,
    OutputCollector<IntWritable, VectorWritable> out, Reporter reporter)
    throws IOException {

  if (!it.hasNext()) {
    return;
  }

  Vector accumulator = new RandomAccessSparseVector(it.next().get());
  while (it.hasNext()) {
    Vector row = it.next().get();
    accumulator.assign(row, Functions.PLUS);
  }

  out.collect(rowNum, new VectorWritable(new SequentialAccessSparseVector(
      accumulator)));
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
protected void map(IntWritable row, VectorWritable vector, Context context) 
  throws IOException, InterruptedException {
  
  // set up the return value and perform the computations
  double norm = vectorNorm(vector.get());
  Vector w = vector.get().assign(Functions.div(norm));
  RandomAccessSparseVector out = new RandomAccessSparseVector(w);
  
  // finally write the output
  context.write(row, new VectorWritable(out));
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
/**
 * Sums the squares of all elements together, then takes the square root
 * of that sum.
 * @param u
 * @return
 */
private static double vectorNorm(Iterable<Vector.Element> u) {
  double retval = 0.0;
  for (Vector.Element e : u) {
    retval += Functions.POW.apply(e.get(), 2);
  }
  return Functions.SQRT.apply(retval);
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
protected void map(IntWritable key, VectorWritable row, Context ctx) 
  throws IOException, InterruptedException {
  
  for (Vector.Element e : row.get()) {
    double dii = Functions.SQRT.apply(diagonal.get(key.get()));
    double djj = Functions.SQRT.apply(diagonal.get(e.index()));
    double mij = e.get();
    e.set(dii * mij * djj);
  }
  ctx.write(key, row);
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
public void update(int termId, Vector topicCounts) {
  for (int x = 0; x < numTopics; x++) {
    Vector v = topicTermCounts.viewRow(x);
    v.set(termId, v.get(termId) + topicCounts.get(x));
  }
  topicSums.assign(topicCounts, Functions.PLUS);
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
protected void reduce(WritableComparable<?> key, Iterable<VectorWritable> values, Context ctx)
  throws IOException, InterruptedException {
  Vector vector = null;
  for (VectorWritable v : values) {
    if (vector == null) {
      vector = v.get();
    } else {
      vector.assign(v.get(), Functions.PLUS);
    }
  }
  ctx.write(key, new VectorWritable(vector));
}
 

import org.apache.mahout.math.function.Functions; //导入依赖的package包/类
@Override
public void reduce(NullWritable n, Iterable<VectorWritable> vectors,
		Context context) throws IOException, InterruptedException {

	/**
	 * Add together partial column-wise sums from mappers
	 */
	for (VectorWritable v : vectors) {
		if (outputVector == null) {
			outputVector = v.get();
		} else {
			outputVector.assign(v.get(), Functions.PLUS);
		}
	}

	/**
	 * Divide total column-wise sum by count of vectors, which
	 * corresponds to the number of rows in the DistributedRowMatrix
	 */
	if (outputVector != null) {
		outputVectorWritable.set(outputVector.viewPart(1,
				outputVector.size() - 1).divide(outputVector.get(0)));
		context.write(ONE, outputVectorWritable);
	} else {
		Vector emptyVector = ClassUtils.instantiateAs(vectorClass,
				Vector.class, new Class<?>[] { int.class },
				new Object[] { 0 });
		context.write(ONE, new VectorWritable(emptyVector));
	}
}