import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
/**
 * This method overrides the Vector.assign method to allow optimization for
 * ZeroIndifferent functions
 * 
 * @param vector
 *          the vector to be updated
 * @param other
 *          the other vector
 * @param function
 *          the function that operates on elements of the two vectors
 * @return the modified vector
 */
static public Vector vectorAssign(Vector vector, Vector other, ZeroIndifferentFunc function) {
  if (vector.size() != other.size()) {
    throw new CardinalityException(vector.size(), other.size());
  }
  // special case: iterate only over the non-zero elements of the vector to
  // add
  Iterator<Element> it = other.nonZeroes().iterator();
  Element e;
  while (it.hasNext() && (e = it.next()) != null) {
    double val = vector.getQuick(e.index());
    double newVal = function.apply(val, e.get());
    vector.setQuick(e.index(), newVal);
  }
  return vector;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private static void printSequenceFile(String inputStr, int printRow) throws IOException {
  Configuration conf = new Configuration();
  Path finalNumberFile = new Path(inputStr);
  SequenceFile.Reader reader = new SequenceFile.Reader(FileSystem.get(conf),
      finalNumberFile, conf);
  IntWritable key = new IntWritable();
  VectorWritable value = new VectorWritable();
  Vector printVector = null;
  while (reader.next(key, value)) {
    if (key.get() == printRow)
      printVector = value.get();
    int cnt = 0;
    Iterator<Element> iter = value.get().nonZeroes().iterator();
    for (; iter.hasNext(); iter.next())
      cnt++;
    System.out.println("# "+ key + " " + cnt + " " + value.get().zSum());
  }
  reader.close();
  if (printVector != null)
    System.out.println("##### "+ printRow + " " + printVector);
  else
    System.out.println("##### "+ key + " " + value.get());
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的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.math.Vector.Element; //导入依赖的package包/类
public Vector train(Vector datapoint) {
	this.initializeAtoms(datapoint);		
	Vector projection = this.transformer.transform(datapoint, this.dictionaryMatrix);

	for (int i=0; i < this.numberOfAtoms; i++) {
		Vector atom = this.dictionaryMatrix.viewColumn(i);
		double projectionWeight = projection.get(i);
		Vector difference = atom.minus(datapoint);
		for (Element elem : difference.nonZeroes()) {
			atom.incrementQuick(elem.index(), - this.learningRate * projectionWeight * elem.get());
		}
	}
	this.regularize();
	this.transformer.clearCaches();		
	
	return projection;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private void regularize() {
	for (int i=0; i < this.numberOfAtoms; i++) {
		Vector atom = this.dictionaryMatrix.viewColumn(i);
		List<Integer> indicesToRemove = new ArrayList<Integer>();
		for (Element elem : atom.nonZeroes()) {
			double regularizedValue = elem.get() - (this.learningRate 
					* (elem.get() 
							* this.l2Penalty
							+ this.l1Penalty * Math.signum(elem.get())));
			if (regularizedValue == 0.0 || Math.abs(regularizedValue) < this.l1Penalty) {
				indicesToRemove.add(elem.index());
			} else {
				atom.setQuick(elem.index(), regularizedValue);
			}
		}
		for (int indexToRemove : indicesToRemove) {
			atom.setQuick(indexToRemove, 0.0);
		}
	}
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
public int printDistributedRowMatrix() {
  System.out.println("RowPath: " + this.rowPath);
  Iterator<MatrixSlice> iterator = this.iterateAll();
  int count = 0;
  while (iterator.hasNext()) {
    MatrixSlice slice = iterator.next();
    Vector v = slice.vector();
    int size = v.size();
    for (int i = 0; i < size; i++) {
      Element e = v.getElement(i);
      count++;
      System.out.print(e.get() + " ");
    }
    System.out.println();
  }
  return count;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
/**
 * Compute principal component analysis where the input is a path for a hadoop sequence File <IntWritable key, VectorWritable value>
 * @param sc 
 * 	 		Spark context that contains the configuration parameters and represents connection to the cluster 
 * 			(used to create RDDs, accumulators and broadcast variables on that cluster)
 * @param inputPath
 * 			Path to the sequence file that represents the input matrix
 * @param nRows
 * 			Number of rows in input Matrix
 * @param nCols
 * 			Number of columns in input Matrix
 * @param nPCs
 * 			Number of desired principal components
 * @param errRate
 * 			The sampling rate that is used for computing the reconstruction error
 * @param maxIterations 
 * 			Maximum number of iterations before terminating
 * @return Matrix of size nCols X nPCs having the desired principal components
 */
public static org.apache.spark.mllib.linalg.Matrix computePrincipalComponents(JavaSparkContext sc, String inputPath, String outputPath, final int nRows, final int nCols, final int nPCs, final double errRate, final int maxIterations, final int computeProjectedMatrix) {
  

//Read from sequence file
JavaPairRDD<IntWritable,VectorWritable> seqVectors = sc.sequenceFile(inputPath, IntWritable.class, VectorWritable.class);
	   
	   
//Convert sequence file to RDD<org.apache.spark.mllib.linalg.Vector> of Vectors
JavaRDD<org.apache.spark.mllib.linalg.Vector> vectors=seqVectors.map(new Function<Tuple2<IntWritable,VectorWritable>, org.apache.spark.mllib.linalg.Vector>() {

              public org.apache.spark.mllib.linalg.Vector call(Tuple2<IntWritable, VectorWritable> arg0)
                              throws Exception {

                org.apache.mahout.math.Vector mahoutVector = arg0._2.get();
                Iterator<Element> elements =  mahoutVector.nonZeroes().iterator();
                ArrayList<Tuple2<Integer, Double>> tupleList = new  ArrayList<Tuple2<Integer, Double>>();
                while(elements.hasNext())
                {
                         Element e = elements.next();
                         if(e.index() >= nCols || e.get() == 0 )
                            continue;
                         Tuple2<Integer,Double> tuple = new Tuple2<Integer,Double>(e.index(), e.get());
                         tupleList.add(tuple);
                }
                org.apache.spark.mllib.linalg.Vector sparkVector = Vectors.sparse(nCols,tupleList);
                return sparkVector;
          }
}).persist(StorageLevel.MEMORY_ONLY_SER());

return computePrincipalComponents(sc, vectors, outputPath, nRows, nCols, nPCs, errRate, maxIterations, computeProjectedMatrix);
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
/**
 * Compute the product(r[i]*SQRT2PI) over all i. Note that the cluster Radius
 * corresponds to the Stdev of a Gaussian and the Center to its Mean.
 */
private void computeProd2piR() {
  zProd2piR = 1.0;
  for (Iterator<Element> it = getRadius().iterateNonZero(); it.hasNext();) {
    Element radius = it.next();
    zProd2piR *= radius.get() * UncommonDistributions.SQRT2PI;
  }
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
/**
 * @param x
 *          a Vector
 * @return the zSum(((x[i]-c[i])/r[i])^2) over all i
 */
private double sumXminusCdivRsquared(Vector x) {
  double result = 0;
  for (Iterator<Element> it = getRadius().iterateNonZero(); it.hasNext();) {
    Element radiusElem = it.next();
    int index = radiusElem.index();
    double quotient = (x.get(index) - getCenter().get(index))
        / radiusElem.get();
    result += quotient * quotient;
  }
  return result;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private static void writeAllAboveThreshold(List<Cluster> clusterModels, Double clusterClassificationThreshold,
    SequenceFile.Writer writer, VectorWritable vw, Vector pdfPerCluster) throws IOException {
  Iterator<Element> iterateNonZero = pdfPerCluster.iterateNonZero();
  while (iterateNonZero.hasNext()) {
    Element pdf = iterateNonZero.next();
    if (pdf.get() >= clusterClassificationThreshold) {
      WeightedVectorWritable wvw = new WeightedVectorWritable(pdf.get(), vw.get());
      int clusterIndex = pdf.index();
      write(clusterModels, writer, wvw, clusterIndex);
    }
  }
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private void writeAllAboveThreshold(VectorWritable vw, Context context,
    Vector pdfPerCluster) throws IOException, InterruptedException {
  Iterator<Element> iterateNonZero = pdfPerCluster.iterateNonZero();
  while (iterateNonZero.hasNext()) {
    Element pdf = iterateNonZero.next();
    if (pdf.get() >= threshold) {
      int clusterIndex = pdf.index();
      write(vw, context, clusterIndex, pdf.get());
    }
  }
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
@Override
protected void map(WritableComparable<?> key, VectorWritable value, Context context) throws IOException,
    InterruptedException {
  Vector probabilities = classifier.classify(value.get());
  Vector selections = policy.select(probabilities);
  for (Iterator<Element> it = selections.iterateNonZero(); it.hasNext();) {
    Element el = it.next();
    classifier.train(el.index(), value.get(), el.get());
  }
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
/**
 *  Math.pow is clever about integer-valued doubles
 */
@Override
public double distance(Vector v1, Vector v2) {
  Vector distVector = v1.minus(v2);
  double sum = 0.0;
  Iterator<Element> it = distVector.iterateNonZero();
  while (it.hasNext()) {
    Element e = it.next();
    sum += Math.pow(Math.abs(e.get()), exponent);
  }
  return Math.pow(sum, 1.0 / exponent);
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
public static double computeAverageNonzeroElementRank(Vector inputVector, Vector ranking) {
    RankMap rankMap = getRankMap(ranking);
    double averageRank = 0.0;
    for (Element elem : inputVector.nonZeroes()) {
        averageRank += rankMap.get(elem.index());
    }
    averageRank = averageRank / inputVector.getNumNonZeroElements();
    return averageRank;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
public static Vector transposedDictionaryTimesDatapoint(SparseColumnMatrix dictionary, Vector datapoint) {
	Vector output = new DenseVector(dictionary.numCols());
	for (int i=0; i < dictionary.numCols(); i++) {
		Vector row = dictionary.viewColumn(i);
		double value = 0.0;
		for (Element elem : row.nonZeroes()) {
			value = value + elem.get() * datapoint.get(elem.index());
		}
		output.setQuick(i, value);
	}
	return output;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
public static Vector inverseTransform(SparseColumnMatrix dictionary, Vector projection) {
	Vector output = new RandomAccessSparseVector(dictionary.numRows());
	for (int i=0; i < dictionary.numCols(); i++) {
		Vector atom = dictionary.viewColumn(i);
		double projectionWeight = projection.get(i);
		for (Element elem : atom.nonZeroes()) {
			output.incrementQuick(elem.index(), elem.get() * projectionWeight);
		}
	}
	return output;
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private static Matrix sparsifyData(Matrix denseData, int cardinality) {
	ArrayList<Vector> vectors = new ArrayList<Vector>();
	for (Vector denseVector : denseData) {
		Vector sparsefiedVector = new RandomAccessSparseVector(cardinality);
		for (Element elem : denseVector.nonZeroes()) {
			sparsefiedVector.set(elem.index(), elem.get());
		}
		vectors.add(sparsefiedVector);
	}
	return new SparseRowMatrix(vectors.size(), cardinality, vectors.toArray(new Vector[3]));
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private void writeAllAboveThreshold(WritableComparable<?> key, Context context,
    Vector pdfPerCluster) throws IOException, InterruptedException {
  for (Element pdf : pdfPerCluster.nonZeroes()) {
    if (pdf.get() >= threshold) {
      int clusterIndex = pdf.index();
      write(key, context, clusterIndex, pdf.get());
    }
  }
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
private static void writeAllAboveThreshold(List<Cluster> clusterModels,
		Double clusterClassificationThreshold, SequenceFile.Writer writer,
		Text key, Vector pdfPerCluster) throws IOException {
	for (Element pdf : pdfPerCluster.nonZeroes()) {
		if (pdf.get() >= clusterClassificationThreshold) {
			int clusterIndex = pdf.index();
			write(clusterModels, writer, key, clusterIndex);
			// nie ma sensu zapisywac wiecej niz 1, bo i tak nie zapisujemy
			// wagi
			break;
		}
	}
}
 

import org.apache.mahout.math.Vector.Element; //导入依赖的package包/类
@Override
public void map(IntWritable key, VectorWritable value, Context context) throws IOException, InterruptedException
{

	Vector similarityMatrixRow = value.get();

	// Remove self similarity
	similarityMatrixRow.set(key.get(), Double.NEGATIVE_INFINITY); // from the equation

	// determine max non-zero element: (==item index)
	List<RecommendationElement> recommendations = new ArrayList<RecommendationElement>();

	// collect non-zero items:
	Iterator<Element> it = similarityMatrixRow.iterateNonZero();
	while (it.hasNext())
	{
		Element e = it.next();
		recommendations.add(new RecommendationElement(e.index(), e.get()));
		// LOG.info("created new recommendation for " + e.index());
	}

	// sorted list of recommendations: now we have an item id, and similarity value:
	Collections.sort(recommendations, new SimilarityComparator());
	LOG.info("sorted: " + recommendations.size());

	int rank = 1;
	Put put = new Put(RowKeys.getRecommendationKey(collection, recommender, key.get()));
	for (RecommendationElement el : recommendations)
	{
		// if (el.getSimilarityValue() > 0)
		// {
		byte[] data = Recommend.getRecommendation(el.getItemIndex(), rank, el.getSimilarityValue());
		put.add(RECOMMENDATION_COLUMN, Bytes.toBytes(rank), data);
		rank++;
		// }
	}
	context.write(new LongWritable(key.get()), put);

}