http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/HebbianUpdater.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/HebbianUpdater.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/HebbianUpdater.java new file mode 100644 index 0000000..2080c3a --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/HebbianUpdater.java @@ -0,0 +1,71 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.decomposer.hebbian; + + +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.function.PlusMult; + +public class HebbianUpdater implements EigenUpdater { + + @Override + public void update(Vector pseudoEigen, + Vector trainingVector, + TrainingState currentState) { + double trainingVectorNorm = trainingVector.norm(2); + int numPreviousEigens = currentState.getNumEigensProcessed(); + if (numPreviousEigens > 0 && currentState.isFirstPass()) { + updateTrainingProjectionsVector(currentState, trainingVector, numPreviousEigens - 1); + } + if (currentState.getActivationDenominatorSquared() == 0 || trainingVectorNorm == 0) { + if (currentState.getActivationDenominatorSquared() == 0) { + pseudoEigen.assign(trainingVector, new PlusMult(1)); + currentState.setHelperVector(currentState.currentTrainingProjection().clone()); + double helperNorm = currentState.getHelperVector().norm(2); + currentState.setActivationDenominatorSquared(trainingVectorNorm * trainingVectorNorm - helperNorm * helperNorm); + } + return; + } + currentState.setActivationNumerator(pseudoEigen.dot(trainingVector)); + currentState.setActivationNumerator( + currentState.getActivationNumerator() + - currentState.getHelperVector().dot(currentState.currentTrainingProjection())); + + double activation = currentState.getActivationNumerator() + / Math.sqrt(currentState.getActivationDenominatorSquared()); + currentState.setActivationDenominatorSquared( + currentState.getActivationDenominatorSquared() + + 2 * activation * currentState.getActivationNumerator() + + activation * activation + * (trainingVector.getLengthSquared() - currentState.currentTrainingProjection().getLengthSquared())); + if (numPreviousEigens > 0) { + currentState.getHelperVector().assign(currentState.currentTrainingProjection(), new PlusMult(activation)); + } + pseudoEigen.assign(trainingVector, new PlusMult(activation)); + } + + private static void updateTrainingProjectionsVector(TrainingState state, + Vector trainingVector, + int previousEigenIndex) { + Vector previousEigen = state.mostRecentEigen(); + Vector currentTrainingVectorProjection = state.currentTrainingProjection(); + double projection = previousEigen.dot(trainingVector); + currentTrainingVectorProjection.set(previousEigenIndex, projection); + } + +}
http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/TrainingState.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/TrainingState.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/TrainingState.java new file mode 100644 index 0000000..af6c2ef --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/hebbian/TrainingState.java @@ -0,0 +1,143 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.decomposer.hebbian; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.mahout.math.DenseVector; +import org.apache.mahout.math.Matrix; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.decomposer.EigenStatus; + +public class TrainingState { + + private Matrix currentEigens; + private int numEigensProcessed; + private List<Double> currentEigenValues; + private Matrix trainingProjections; + private int trainingIndex; + private Vector helperVector; + private boolean firstPass; + private List<EigenStatus> statusProgress; + private double activationNumerator; + private double activationDenominatorSquared; + + TrainingState(Matrix eigens, Matrix projections) { + currentEigens = eigens; + trainingProjections = projections; + trainingIndex = 0; + helperVector = new DenseVector(eigens.numRows()); + firstPass = true; + statusProgress = new ArrayList<>(); + activationNumerator = 0; + activationDenominatorSquared = 0; + numEigensProcessed = 0; + } + + public Vector mostRecentEigen() { + return currentEigens.viewRow(numEigensProcessed - 1); + } + + public Vector currentTrainingProjection() { + if (trainingProjections.viewRow(trainingIndex) == null) { + trainingProjections.assignRow(trainingIndex, new DenseVector(currentEigens.numCols())); + } + return trainingProjections.viewRow(trainingIndex); + } + + public Matrix getCurrentEigens() { + return currentEigens; + } + + public void setCurrentEigens(Matrix currentEigens) { + this.currentEigens = currentEigens; + } + + public int getNumEigensProcessed() { + return numEigensProcessed; + } + + public void setNumEigensProcessed(int numEigensProcessed) { + this.numEigensProcessed = numEigensProcessed; + } + + public List<Double> getCurrentEigenValues() { + return currentEigenValues; + } + + public void setCurrentEigenValues(List<Double> currentEigenValues) { + this.currentEigenValues = currentEigenValues; + } + + public Matrix getTrainingProjections() { + return trainingProjections; + } + + public void setTrainingProjections(Matrix trainingProjections) { + this.trainingProjections = trainingProjections; + } + + public int getTrainingIndex() { + return trainingIndex; + } + + public void setTrainingIndex(int trainingIndex) { + this.trainingIndex = trainingIndex; + } + + public Vector getHelperVector() { + return helperVector; + } + + public void setHelperVector(Vector helperVector) { + this.helperVector = helperVector; + } + + public boolean isFirstPass() { + return firstPass; + } + + public void setFirstPass(boolean firstPass) { + this.firstPass = firstPass; + } + + public List<EigenStatus> getStatusProgress() { + return statusProgress; + } + + public void setStatusProgress(List<EigenStatus> statusProgress) { + this.statusProgress = statusProgress; + } + + public double getActivationNumerator() { + return activationNumerator; + } + + public void setActivationNumerator(double activationNumerator) { + this.activationNumerator = activationNumerator; + } + + public double getActivationDenominatorSquared() { + return activationDenominatorSquared; + } + + public void setActivationDenominatorSquared(double activationDenominatorSquared) { + this.activationDenominatorSquared = activationDenominatorSquared; + } +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosSolver.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosSolver.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosSolver.java new file mode 100644 index 0000000..61a77db --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosSolver.java @@ -0,0 +1,213 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.decomposer.lanczos; + + +import java.util.EnumMap; +import java.util.Map; + +import com.google.common.base.Preconditions; +import org.apache.mahout.math.Matrix; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorIterable; +import org.apache.mahout.math.function.DoubleFunction; +import org.apache.mahout.math.function.PlusMult; +import org.apache.mahout.math.solver.EigenDecomposition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Simple implementation of the <a href="http://en.wikipedia.org/wiki/Lanczos_algorithm";>Lanczos algorithm</a> for + * finding eigenvalues of a symmetric matrix, applied to non-symmetric matrices by applying Matrix.timesSquared(vector) + * as the "matrix-multiplication" method.<p> + * + * See the SSVD code for a better option + * {@link org.apache.mahout.math.ssvd.SequentialBigSvd} + * See also the docs on + * <a href=https://mahout.apache.org/users/dim-reduction/ssvd.html>stochastic + * projection SVD</a> + * <p> + * To avoid floating point overflow problems which arise in power-methods like Lanczos, an initial pass is made + * through the input matrix to + * <ul> + * <li>generate a good starting seed vector by summing all the rows of the input matrix, and</li> + * <li>compute the trace(inputMatrix<sup>t</sup>*matrix) + * </ul> + * <p> + * This latter value, being the sum of all of the singular values, is used to rescale the entire matrix, effectively + * forcing the largest singular value to be strictly less than one, and transforming floating point <em>overflow</em> + * problems into floating point <em>underflow</em> (ie, very small singular values will become invisible, as they + * will appear to be zero and the algorithm will terminate). + * <p>This implementation uses {@link EigenDecomposition} to do the + * eigenvalue extraction from the small (desiredRank x desiredRank) tridiagonal matrix. Numerical stability is + * achieved via brute-force: re-orthogonalization against all previous eigenvectors is computed after every pass. + * This can be made smarter if (when!) this proves to be a major bottleneck. Of course, this step can be parallelized + * as well. + * @see org.apache.mahout.math.ssvd.SequentialBigSvd + */ +@Deprecated +public class LanczosSolver { + + private static final Logger log = LoggerFactory.getLogger(LanczosSolver.class); + + public static final double SAFE_MAX = 1.0e150; + + public enum TimingSection { + ITERATE, ORTHOGANLIZE, TRIDIAG_DECOMP, FINAL_EIGEN_CREATE + } + + private final Map<TimingSection, Long> startTimes = new EnumMap<>(TimingSection.class); + private final Map<TimingSection, Long> times = new EnumMap<>(TimingSection.class); + + private static final class Scale extends DoubleFunction { + private final double d; + + private Scale(double d) { + this.d = d; + } + + @Override + public double apply(double arg1) { + return arg1 * d; + } + } + + public void solve(LanczosState state, + int desiredRank) { + solve(state, desiredRank, false); + } + + public void solve(LanczosState state, + int desiredRank, + boolean isSymmetric) { + VectorIterable corpus = state.getCorpus(); + log.info("Finding {} singular vectors of matrix with {} rows, via Lanczos", + desiredRank, corpus.numRows()); + int i = state.getIterationNumber(); + Vector currentVector = state.getBasisVector(i - 1); + Vector previousVector = state.getBasisVector(i - 2); + double beta = 0; + Matrix triDiag = state.getDiagonalMatrix(); + while (i < desiredRank) { + startTime(TimingSection.ITERATE); + Vector nextVector = isSymmetric ? corpus.times(currentVector) : corpus.timesSquared(currentVector); + log.info("{} passes through the corpus so far...", i); + if (state.getScaleFactor() <= 0) { + state.setScaleFactor(calculateScaleFactor(nextVector)); + } + nextVector.assign(new Scale(1.0 / state.getScaleFactor())); + if (previousVector != null) { + nextVector.assign(previousVector, new PlusMult(-beta)); + } + // now orthogonalize + double alpha = currentVector.dot(nextVector); + nextVector.assign(currentVector, new PlusMult(-alpha)); + endTime(TimingSection.ITERATE); + startTime(TimingSection.ORTHOGANLIZE); + orthoganalizeAgainstAllButLast(nextVector, state); + endTime(TimingSection.ORTHOGANLIZE); + // and normalize + beta = nextVector.norm(2); + if (outOfRange(beta) || outOfRange(alpha)) { + log.warn("Lanczos parameters out of range: alpha = {}, beta = {}. Bailing out early!", + alpha, beta); + break; + } + nextVector.assign(new Scale(1 / beta)); + state.setBasisVector(i, nextVector); + previousVector = currentVector; + currentVector = nextVector; + // save the projections and norms! + triDiag.set(i - 1, i - 1, alpha); + if (i < desiredRank - 1) { + triDiag.set(i - 1, i, beta); + triDiag.set(i, i - 1, beta); + } + state.setIterationNumber(++i); + } + startTime(TimingSection.TRIDIAG_DECOMP); + + log.info("Lanczos iteration complete - now to diagonalize the tri-diagonal auxiliary matrix."); + // at this point, have tridiag all filled out, and basis is all filled out, and orthonormalized + EigenDecomposition decomp = new EigenDecomposition(triDiag); + + Matrix eigenVects = decomp.getV(); + Vector eigenVals = decomp.getRealEigenvalues(); + endTime(TimingSection.TRIDIAG_DECOMP); + startTime(TimingSection.FINAL_EIGEN_CREATE); + for (int row = 0; row < i; row++) { + Vector realEigen = null; + + Vector ejCol = eigenVects.viewColumn(row); + int size = Math.min(ejCol.size(), state.getBasisSize()); + for (int j = 0; j < size; j++) { + double d = ejCol.get(j); + Vector rowJ = state.getBasisVector(j); + if (realEigen == null) { + realEigen = rowJ.like(); + } + realEigen.assign(rowJ, new PlusMult(d)); + } + + Preconditions.checkState(realEigen != null); + assert realEigen != null; + + realEigen = realEigen.normalize(); + state.setRightSingularVector(row, realEigen); + double e = eigenVals.get(row) * state.getScaleFactor(); + if (!isSymmetric) { + e = Math.sqrt(e); + } + log.info("Eigenvector {} found with eigenvalue {}", row, e); + state.setSingularValue(row, e); + } + log.info("LanczosSolver finished."); + endTime(TimingSection.FINAL_EIGEN_CREATE); + } + + protected static double calculateScaleFactor(Vector nextVector) { + return nextVector.norm(2); + } + + private static boolean outOfRange(double d) { + return Double.isNaN(d) || d > SAFE_MAX || -d > SAFE_MAX; + } + + protected static void orthoganalizeAgainstAllButLast(Vector nextVector, LanczosState state) { + for (int i = 0; i < state.getIterationNumber(); i++) { + Vector basisVector = state.getBasisVector(i); + double alpha; + if (basisVector == null || (alpha = nextVector.dot(basisVector)) == 0.0) { + continue; + } + nextVector.assign(basisVector, new PlusMult(-alpha)); + } + } + + private void startTime(TimingSection section) { + startTimes.put(section, System.nanoTime()); + } + + private void endTime(TimingSection section) { + if (!times.containsKey(section)) { + times.put(section, 0L); + } + times.put(section, times.get(section) + System.nanoTime() - startTimes.get(section)); + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosState.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosState.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosState.java new file mode 100644 index 0000000..2ba34bd --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosState.java @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.decomposer.lanczos; + +import com.google.common.collect.Maps; +import org.apache.mahout.math.DenseMatrix; +import org.apache.mahout.math.Matrix; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorIterable; + +import java.util.Map; + +@Deprecated +public class LanczosState { + + protected Matrix diagonalMatrix; + protected final VectorIterable corpus; + protected double scaleFactor; + protected int iterationNumber; + protected final int desiredRank; + protected Map<Integer, Vector> basis; + protected final Map<Integer, Double> singularValues; + protected Map<Integer, Vector> singularVectors; + + public LanczosState(VectorIterable corpus, int desiredRank, Vector initialVector) { + this.corpus = corpus; + this.desiredRank = desiredRank; + intitializeBasisAndSingularVectors(); + setBasisVector(0, initialVector); + scaleFactor = 0; + diagonalMatrix = new DenseMatrix(desiredRank, desiredRank); + singularValues = Maps.newHashMap(); + iterationNumber = 1; + } + + private void intitializeBasisAndSingularVectors() { + basis = Maps.newHashMap(); + singularVectors = Maps.newHashMap(); + } + + public Matrix getDiagonalMatrix() { + return diagonalMatrix; + } + + public int getIterationNumber() { + return iterationNumber; + } + + public double getScaleFactor() { + return scaleFactor; + } + + public VectorIterable getCorpus() { + return corpus; + } + + public Vector getRightSingularVector(int i) { + return singularVectors.get(i); + } + + public Double getSingularValue(int i) { + return singularValues.get(i); + } + + public Vector getBasisVector(int i) { + return basis.get(i); + } + + public int getBasisSize() { + return basis.size(); + } + + public void setBasisVector(int i, Vector basisVector) { + basis.put(i, basisVector); + } + + public void setScaleFactor(double scale) { + scaleFactor = scale; + } + + public void setIterationNumber(int i) { + iterationNumber = i; + } + + public void setRightSingularVector(int i, Vector vector) { + singularVectors.put(i, vector); + } + + public void setSingularValue(int i, double value) { + singularValues.put(i, value); + } +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java new file mode 100644 index 0000000..de5e216 --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java @@ -0,0 +1,390 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.hadoop; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import java.util.Iterator; + +import org.apache.hadoop.conf.Configurable; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.WritableComparable; +import org.apache.hadoop.mapred.JobClient; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapreduce.Job; +import org.apache.mahout.common.Pair; +import org.apache.mahout.common.iterator.sequencefile.PathFilters; +import org.apache.mahout.common.iterator.sequencefile.PathType; +import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirIterator; +import org.apache.mahout.math.CardinalityException; +import org.apache.mahout.math.MatrixSlice; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorIterable; +import org.apache.mahout.math.VectorWritable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Function; +import com.google.common.collect.Iterators; + +/** + * DistributedRowMatrix is a FileSystem-backed VectorIterable in which the vectors live in a + * SequenceFile<WritableComparable,VectorWritable>, and distributed operations are executed as M/R passes on + * Hadoop. The usage is as follows: <p> + * <p> + * <pre> + * // the path must already contain an already created SequenceFile! + * DistributedRowMatrix m = new DistributedRowMatrix("path/to/vector/sequenceFile", "tmp/path", 10000000, 250000); + * m.setConf(new Configuration()); + * // now if we want to multiply a vector by this matrix, it's dimension must equal the row dimension of this + * // matrix. If we want to timesSquared() a vector by this matrix, its dimension must equal the column dimension + * // of the matrix. + * Vector v = new DenseVector(250000); + * // now the following operation will be done via a M/R pass via Hadoop. + * Vector w = m.timesSquared(v); + * </pre> + * + */ +public class DistributedRowMatrix implements VectorIterable, Configurable { + public static final String KEEP_TEMP_FILES = "DistributedMatrix.keep.temp.files"; + + private static final Logger log = LoggerFactory.getLogger(DistributedRowMatrix.class); + + private final Path inputPath; + private final Path outputTmpPath; + private Configuration conf; + private Path rowPath; + private Path outputTmpBasePath; + private final int numRows; + private final int numCols; + private boolean keepTempFiles; + + public DistributedRowMatrix(Path inputPath, + Path outputTmpPath, + int numRows, + int numCols) { + this(inputPath, outputTmpPath, numRows, numCols, false); + } + + public DistributedRowMatrix(Path inputPath, + Path outputTmpPath, + int numRows, + int numCols, + boolean keepTempFiles) { + this.inputPath = inputPath; + this.outputTmpPath = outputTmpPath; + this.numRows = numRows; + this.numCols = numCols; + this.keepTempFiles = keepTempFiles; + } + + @Override + public Configuration getConf() { + return conf; + } + + @Override + public void setConf(Configuration conf) { + this.conf = conf; + try { + FileSystem fs = FileSystem.get(inputPath.toUri(), conf); + rowPath = fs.makeQualified(inputPath); + outputTmpBasePath = fs.makeQualified(outputTmpPath); + keepTempFiles = conf.getBoolean(KEEP_TEMP_FILES, false); + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + } + + public Path getRowPath() { + return rowPath; + } + + public Path getOutputTempPath() { + return outputTmpBasePath; + } + + public void setOutputTempPathString(String outPathString) { + try { + outputTmpBasePath = FileSystem.get(conf).makeQualified(new Path(outPathString)); + } catch (IOException ioe) { + log.warn("Unable to set outputBasePath to {}, leaving as {}", + outPathString, outputTmpBasePath); + } + } + + @Override + public Iterator<MatrixSlice> iterateNonEmpty() { + return iterator(); + } + + @Override + public Iterator<MatrixSlice> iterateAll() { + try { + Path pathPattern = rowPath; + if (FileSystem.get(conf).getFileStatus(rowPath).isDir()) { + pathPattern = new Path(rowPath, "*"); + } + return Iterators.transform( + new SequenceFileDirIterator<IntWritable,VectorWritable>(pathPattern, + PathType.GLOB, + PathFilters.logsCRCFilter(), + null, + true, + conf), + new Function<Pair<IntWritable,VectorWritable>,MatrixSlice>() { + @Override + public MatrixSlice apply(Pair<IntWritable, VectorWritable> from) { + return new MatrixSlice(from.getSecond().get(), from.getFirst().get()); + } + }); + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + } + + @Override + public int numSlices() { + return numRows(); + } + + @Override + public int numRows() { + return numRows; + } + + @Override + public int numCols() { + return numCols; + } + + + /** + * This implements matrix this.transpose().times(other) + * @param other a DistributedRowMatrix + * @return a DistributedRowMatrix containing the product + */ + public DistributedRowMatrix times(DistributedRowMatrix other) throws IOException { + return times(other, new Path(outputTmpBasePath.getParent(), "productWith-" + (System.nanoTime() & 0xFF))); + } + + /** + * This implements matrix this.transpose().times(other) + * @param other a DistributedRowMatrix + * @param outPath path to write result to + * @return a DistributedRowMatrix containing the product + */ + public DistributedRowMatrix times(DistributedRowMatrix other, Path outPath) throws IOException { + if (numRows != other.numRows()) { + throw new CardinalityException(numRows, other.numRows()); + } + + Configuration initialConf = getConf() == null ? new Configuration() : getConf(); + Configuration conf = + MatrixMultiplicationJob.createMatrixMultiplyJobConf(initialConf, + rowPath, + other.rowPath, + outPath, + other.numCols); + JobClient.runJob(new JobConf(conf)); + DistributedRowMatrix out = new DistributedRowMatrix(outPath, outputTmpPath, numCols, other.numCols()); + out.setConf(conf); + return out; + } + + public Vector columnMeans() throws IOException { + return columnMeans("SequentialAccessSparseVector"); + } + + /** + * Returns the column-wise mean of a DistributedRowMatrix + * + * @param vectorClass + * desired class for the column-wise mean vector e.g. + * RandomAccessSparseVector, DenseVector + * @return Vector containing the column-wise mean of this + */ + public Vector columnMeans(String vectorClass) throws IOException { + Path outputVectorTmpPath = + new Path(outputTmpBasePath, new Path(Long.toString(System.nanoTime()))); + Configuration initialConf = + getConf() == null ? new Configuration() : getConf(); + String vectorClassFull = "org.apache.mahout.math." + vectorClass; + Vector mean = MatrixColumnMeansJob.run(initialConf, rowPath, outputVectorTmpPath, vectorClassFull); + if (!keepTempFiles) { + FileSystem fs = outputVectorTmpPath.getFileSystem(conf); + fs.delete(outputVectorTmpPath, true); + } + return mean; + } + + public DistributedRowMatrix transpose() throws IOException { + Path outputPath = new Path(rowPath.getParent(), "transpose-" + (System.nanoTime() & 0xFF)); + Configuration initialConf = getConf() == null ? new Configuration() : getConf(); + Job transposeJob = TransposeJob.buildTransposeJob(initialConf, rowPath, outputPath, numRows); + + try { + transposeJob.waitForCompletion(true); + } catch (Exception e) { + throw new IllegalStateException("transposition failed", e); + } + + DistributedRowMatrix m = new DistributedRowMatrix(outputPath, outputTmpPath, numCols, numRows); + m.setConf(this.conf); + return m; + } + + @Override + public Vector times(Vector v) { + try { + Configuration initialConf = getConf() == null ? new Configuration() : getConf(); + Path outputVectorTmpPath = new Path(outputTmpBasePath, new Path(Long.toString(System.nanoTime()))); + + Job job = TimesSquaredJob.createTimesJob(initialConf, v, numRows, rowPath, outputVectorTmpPath); + + try { + job.waitForCompletion(true); + } catch (Exception e) { + throw new IllegalStateException("times failed", e); + } + + Vector result = TimesSquaredJob.retrieveTimesSquaredOutputVector(outputVectorTmpPath, conf); + if (!keepTempFiles) { + FileSystem fs = outputVectorTmpPath.getFileSystem(conf); + fs.delete(outputVectorTmpPath, true); + } + return result; + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + } + + @Override + public Vector timesSquared(Vector v) { + try { + Configuration initialConf = getConf() == null ? new Configuration() : getConf(); + Path outputVectorTmpPath = new Path(outputTmpBasePath, new Path(Long.toString(System.nanoTime()))); + + Job job = TimesSquaredJob.createTimesSquaredJob(initialConf, v, rowPath, outputVectorTmpPath); + + try { + job.waitForCompletion(true); + } catch (Exception e) { + throw new IllegalStateException("timesSquared failed", e); + } + + Vector result = TimesSquaredJob.retrieveTimesSquaredOutputVector(outputVectorTmpPath, conf); + if (!keepTempFiles) { + FileSystem fs = outputVectorTmpPath.getFileSystem(conf); + fs.delete(outputVectorTmpPath, true); + } + return result; + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + } + + @Override + public Iterator<MatrixSlice> iterator() { + return iterateAll(); + } + + public static class MatrixEntryWritable implements WritableComparable<MatrixEntryWritable> { + private int row; + private int col; + private double val; + + public int getRow() { + return row; + } + + public void setRow(int row) { + this.row = row; + } + + public int getCol() { + return col; + } + + public void setCol(int col) { + this.col = col; + } + + public double getVal() { + return val; + } + + public void setVal(double val) { + this.val = val; + } + + @Override + public int compareTo(MatrixEntryWritable o) { + if (row > o.row) { + return 1; + } else if (row < o.row) { + return -1; + } else { + if (col > o.col) { + return 1; + } else if (col < o.col) { + return -1; + } else { + return 0; + } + } + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof MatrixEntryWritable)) { + return false; + } + MatrixEntryWritable other = (MatrixEntryWritable) o; + return row == other.row && col == other.col; + } + + @Override + public int hashCode() { + return row + 31 * col; + } + + @Override + public void write(DataOutput out) throws IOException { + out.writeInt(row); + out.writeInt(col); + out.writeDouble(val); + } + + @Override + public void readFields(DataInput in) throws IOException { + row = in.readInt(); + col = in.readInt(); + val = in.readDouble(); + } + + @Override + public String toString() { + return "(" + row + ',' + col + "):" + val; + } + } +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixColumnMeansJob.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixColumnMeansJob.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixColumnMeansJob.java new file mode 100644 index 0000000..b4f459a --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixColumnMeansJob.java @@ -0,0 +1,236 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package org.apache.mahout.math.hadoop; + +import java.io.IOException; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.NullWritable; +import org.apache.hadoop.io.Writable; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.mapreduce.Reducer; +import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; +import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; +import org.apache.mahout.common.ClassUtils; +import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterator; +import org.apache.mahout.math.DenseVector; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorWritable; +import org.apache.mahout.math.function.Functions; + +import com.google.common.io.Closeables; + +/** + * MatrixColumnMeansJob is a job for calculating the column-wise mean of a + * DistributedRowMatrix. This job can be accessed using + * DistributedRowMatrix.columnMeans() + */ +public final class MatrixColumnMeansJob { + + public static final String VECTOR_CLASS = + "DistributedRowMatrix.columnMeans.vector.class"; + + private MatrixColumnMeansJob() { + } + + public static Vector run(Configuration conf, + Path inputPath, + Path outputVectorTmpPath) throws IOException { + return run(conf, inputPath, outputVectorTmpPath, null); + } + + /** + * Job for calculating column-wise mean of a DistributedRowMatrix + * + * @param initialConf + * @param inputPath + * path to DistributedRowMatrix input + * @param outputVectorTmpPath + * path for temporary files created during job + * @param vectorClass + * String of desired class for returned vector e.g. DenseVector, + * RandomAccessSparseVector (may be null for {@link DenseVector} ) + * @return Vector containing column-wise mean of DistributedRowMatrix + */ + public static Vector run(Configuration initialConf, + Path inputPath, + Path outputVectorTmpPath, + String vectorClass) throws IOException { + + try { + initialConf.set(VECTOR_CLASS, + vectorClass == null ? DenseVector.class.getName() + : vectorClass); + + Job job = new Job(initialConf, "MatrixColumnMeansJob"); + job.setJarByClass(MatrixColumnMeansJob.class); + + FileOutputFormat.setOutputPath(job, outputVectorTmpPath); + + outputVectorTmpPath.getFileSystem(job.getConfiguration()) + .delete(outputVectorTmpPath, true); + job.setNumReduceTasks(1); + FileOutputFormat.setOutputPath(job, outputVectorTmpPath); + FileInputFormat.addInputPath(job, inputPath); + job.setInputFormatClass(SequenceFileInputFormat.class); + job.setOutputFormatClass(SequenceFileOutputFormat.class); + FileOutputFormat.setOutputPath(job, outputVectorTmpPath); + + job.setMapperClass(MatrixColumnMeansMapper.class); + job.setReducerClass(MatrixColumnMeansReducer.class); + job.setMapOutputKeyClass(NullWritable.class); + job.setMapOutputValueClass(VectorWritable.class); + job.setOutputKeyClass(IntWritable.class); + job.setOutputValueClass(VectorWritable.class); + job.submit(); + job.waitForCompletion(true); + + Path tmpFile = new Path(outputVectorTmpPath, "part-r-00000"); + SequenceFileValueIterator<VectorWritable> iterator = + new SequenceFileValueIterator<>(tmpFile, true, initialConf); + try { + if (iterator.hasNext()) { + return iterator.next().get(); + } else { + return (Vector) Class.forName(vectorClass).getConstructor(int.class) + .newInstance(0); + } + } finally { + Closeables.close(iterator, true); + } + } catch (IOException ioe) { + throw ioe; + } catch (Throwable thr) { + throw new IOException(thr); + } + } + + /** + * Mapper for calculation of column-wise mean. + */ + public static class MatrixColumnMeansMapper extends + Mapper<Writable, VectorWritable, NullWritable, VectorWritable> { + + private Vector runningSum; + private String vectorClass; + + @Override + public void setup(Context context) { + vectorClass = context.getConfiguration().get(VECTOR_CLASS); + } + + /** + * 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); + } + } + + /** + * The column-wise sum is written at the cleanup stage. A single reducer is + * forced so null can be used for the key + */ + @Override + public void cleanup(Context context) throws InterruptedException, + IOException { + if (runningSum != null) { + context.write(NullWritable.get(), new VectorWritable(runningSum)); + } + } + + } + + /** + * The reducer adds the partial column-wise sums from each of the mappers to + * compute the total column-wise sum. The total sum is then divided by the + * total count of vectors to determine the column-wise mean. + */ + public static class MatrixColumnMeansReducer extends + Reducer<NullWritable, VectorWritable, IntWritable, VectorWritable> { + + private static final IntWritable ONE = new IntWritable(1); + + private String vectorClass; + private Vector outputVector; + private final VectorWritable outputVectorWritable = new VectorWritable(); + + @Override + public void setup(Context context) { + vectorClass = context.getConfiguration().get(VECTOR_CLASS); + } + + @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)); + } + } + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java new file mode 100644 index 0000000..48eda08 --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java @@ -0,0 +1,177 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.hadoop; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.mapred.FileOutputFormat; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.MapReduceBase; +import org.apache.hadoop.mapred.Mapper; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.Reducer; +import org.apache.hadoop.mapred.Reporter; +import org.apache.hadoop.mapred.SequenceFileInputFormat; +import org.apache.hadoop.mapred.SequenceFileOutputFormat; +import org.apache.hadoop.mapred.join.CompositeInputFormat; +import org.apache.hadoop.mapred.join.TupleWritable; +import org.apache.hadoop.util.ToolRunner; +import org.apache.mahout.common.AbstractJob; +import org.apache.mahout.math.RandomAccessSparseVector; +import org.apache.mahout.math.SequentialAccessSparseVector; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorWritable; +import org.apache.mahout.math.function.Functions; + +import java.io.IOException; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +/** + * This still uses the old MR api and as with all things in Mahout that are MapReduce is now part of 'mahout-mr'. + * There is no plan to convert the old MR api used here to the new MR api. + * This will be replaced by the new Spark based Linear Algebra bindings. + */ + +public class MatrixMultiplicationJob extends AbstractJob { + + private static final String OUT_CARD = "output.vector.cardinality"; + + public static Configuration createMatrixMultiplyJobConf(Path aPath, + Path bPath, + Path outPath, + int outCardinality) { + return createMatrixMultiplyJobConf(new Configuration(), aPath, bPath, outPath, outCardinality); + } + + public static Configuration createMatrixMultiplyJobConf(Configuration initialConf, + Path aPath, + Path bPath, + Path outPath, + int outCardinality) { + JobConf conf = new JobConf(initialConf, MatrixMultiplicationJob.class); + conf.setInputFormat(CompositeInputFormat.class); + conf.set("mapred.join.expr", CompositeInputFormat.compose( + "inner", SequenceFileInputFormat.class, aPath, bPath)); + conf.setInt(OUT_CARD, outCardinality); + conf.setOutputFormat(SequenceFileOutputFormat.class); + FileOutputFormat.setOutputPath(conf, outPath); + conf.setMapperClass(MatrixMultiplyMapper.class); + conf.setCombinerClass(MatrixMultiplicationReducer.class); + conf.setReducerClass(MatrixMultiplicationReducer.class); + conf.setMapOutputKeyClass(IntWritable.class); + conf.setMapOutputValueClass(VectorWritable.class); + conf.setOutputKeyClass(IntWritable.class); + conf.setOutputValueClass(VectorWritable.class); + return conf; + } + + public static void main(String[] args) throws Exception { + ToolRunner.run(new MatrixMultiplicationJob(), args); + } + + @Override + public int run(String[] strings) throws Exception { + addOption("numRowsA", "nra", "Number of rows of the first input matrix", true); + addOption("numColsA", "nca", "Number of columns of the first input matrix", true); + addOption("numRowsB", "nrb", "Number of rows of the second input matrix", true); + + addOption("numColsB", "ncb", "Number of columns of the second input matrix", true); + addOption("inputPathA", "ia", "Path to the first input matrix", true); + addOption("inputPathB", "ib", "Path to the second input matrix", true); + + addOption("outputPath", "op", "Path to the output matrix", false); + + Map<String, List<String>> argMap = parseArguments(strings); + if (argMap == null) { + return -1; + } + + DistributedRowMatrix a = new DistributedRowMatrix(new Path(getOption("inputPathA")), + new Path(getOption("tempDir")), + Integer.parseInt(getOption("numRowsA")), + Integer.parseInt(getOption("numColsA"))); + DistributedRowMatrix b = new DistributedRowMatrix(new Path(getOption("inputPathB")), + new Path(getOption("tempDir")), + Integer.parseInt(getOption("numRowsB")), + Integer.parseInt(getOption("numColsB"))); + + a.setConf(new Configuration(getConf())); + b.setConf(new Configuration(getConf())); + + if (hasOption("outputPath")) { + a.times(b, new Path(getOption("outputPath"))); + } else { + a.times(b); + } + + return 0; + } + + public static class MatrixMultiplyMapper extends MapReduceBase + implements Mapper<IntWritable,TupleWritable,IntWritable,VectorWritable> { + + private int outCardinality; + private final IntWritable row = new IntWritable(); + + @Override + public void configure(JobConf conf) { + outCardinality = conf.getInt(OUT_CARD, Integer.MAX_VALUE); + } + + @Override + public void map(IntWritable index, + TupleWritable v, + OutputCollector<IntWritable,VectorWritable> out, + Reporter reporter) throws IOException { + boolean firstIsOutFrag = ((VectorWritable)v.get(0)).get().size() == outCardinality; + Vector outFrag = firstIsOutFrag ? ((VectorWritable)v.get(0)).get() : ((VectorWritable)v.get(1)).get(); + Vector multiplier = firstIsOutFrag ? ((VectorWritable)v.get(1)).get() : ((VectorWritable)v.get(0)).get(); + + VectorWritable outVector = new VectorWritable(); + for (Vector.Element e : multiplier.nonZeroes()) { + row.set(e.index()); + outVector.set(outFrag.times(e.get())); + out.collect(row, outVector); + } + } + } + + public static class MatrixMultiplicationReducer extends MapReduceBase + implements Reducer<IntWritable,VectorWritable,IntWritable,VectorWritable> { + + @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))); + } + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java new file mode 100644 index 0000000..e234eb9 --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java @@ -0,0 +1,251 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.hadoop; + +import com.google.common.io.Closeables; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.filecache.DistributedCache; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.NullWritable; +import org.apache.hadoop.io.SequenceFile; +import org.apache.hadoop.io.WritableComparable; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.mapreduce.Reducer; +import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; +import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; +import org.apache.mahout.common.HadoopUtil; +import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterator; +import org.apache.mahout.math.DenseVector; +import org.apache.mahout.math.RandomAccessSparseVector; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorWritable; +import org.apache.mahout.math.function.Functions; + +import com.google.common.base.Preconditions; + +import java.io.IOException; +import java.net.URI; + +public final class TimesSquaredJob { + + public static final String INPUT_VECTOR = "DistributedMatrix.times.inputVector"; + public static final String IS_SPARSE_OUTPUT = "DistributedMatrix.times.outputVector.sparse"; + public static final String OUTPUT_VECTOR_DIMENSION = "DistributedMatrix.times.output.dimension"; + + public static final String OUTPUT_VECTOR_FILENAME = "DistributedMatrix.times.outputVector"; + + private TimesSquaredJob() { } + + public static Job createTimesSquaredJob(Vector v, Path matrixInputPath, Path outputVectorPath) + throws IOException { + return createTimesSquaredJob(new Configuration(), v, matrixInputPath, outputVectorPath); + } + + public static Job createTimesSquaredJob(Configuration initialConf, Vector v, Path matrixInputPath, + Path outputVectorPath) throws IOException { + + return createTimesSquaredJob(initialConf, v, matrixInputPath, outputVectorPath, TimesSquaredMapper.class, + VectorSummingReducer.class); + } + + public static Job createTimesJob(Vector v, int outDim, Path matrixInputPath, Path outputVectorPath) + throws IOException { + + return createTimesJob(new Configuration(), v, outDim, matrixInputPath, outputVectorPath); + } + + public static Job createTimesJob(Configuration initialConf, Vector v, int outDim, Path matrixInputPath, + Path outputVectorPath) throws IOException { + + return createTimesSquaredJob(initialConf, v, outDim, matrixInputPath, outputVectorPath, TimesMapper.class, + VectorSummingReducer.class); + } + + public static Job createTimesSquaredJob(Vector v, Path matrixInputPath, Path outputVectorPathBase, + Class<? extends TimesSquaredMapper> mapClass, Class<? extends VectorSummingReducer> redClass) throws IOException { + + return createTimesSquaredJob(new Configuration(), v, matrixInputPath, outputVectorPathBase, mapClass, redClass); + } + + public static Job createTimesSquaredJob(Configuration initialConf, Vector v, Path matrixInputPath, + Path outputVectorPathBase, Class<? extends TimesSquaredMapper> mapClass, + Class<? extends VectorSummingReducer> redClass) throws IOException { + + return createTimesSquaredJob(initialConf, v, v.size(), matrixInputPath, outputVectorPathBase, mapClass, redClass); + } + + public static Job createTimesSquaredJob(Vector v, int outputVectorDim, Path matrixInputPath, + Path outputVectorPathBase, Class<? extends TimesSquaredMapper> mapClass, + Class<? extends VectorSummingReducer> redClass) throws IOException { + + return createTimesSquaredJob(new Configuration(), v, outputVectorDim, matrixInputPath, outputVectorPathBase, + mapClass, redClass); + } + + public static Job createTimesSquaredJob(Configuration initialConf, Vector v, int outputVectorDim, + Path matrixInputPath, Path outputVectorPathBase, Class<? extends TimesSquaredMapper> mapClass, + Class<? extends VectorSummingReducer> redClass) throws IOException { + + FileSystem fs = FileSystem.get(matrixInputPath.toUri(), initialConf); + matrixInputPath = fs.makeQualified(matrixInputPath); + outputVectorPathBase = fs.makeQualified(outputVectorPathBase); + + long now = System.nanoTime(); + Path inputVectorPath = new Path(outputVectorPathBase, INPUT_VECTOR + '/' + now); + + + SequenceFile.Writer inputVectorPathWriter = null; + + try { + inputVectorPathWriter = new SequenceFile.Writer(fs, initialConf, inputVectorPath, NullWritable.class, + VectorWritable.class); + inputVectorPathWriter.append(NullWritable.get(), new VectorWritable(v)); + } finally { + Closeables.close(inputVectorPathWriter, false); + } + + URI ivpURI = inputVectorPath.toUri(); + DistributedCache.setCacheFiles(new URI[] { ivpURI }, initialConf); + + Job job = HadoopUtil.prepareJob(matrixInputPath, new Path(outputVectorPathBase, OUTPUT_VECTOR_FILENAME), + SequenceFileInputFormat.class, mapClass, NullWritable.class, VectorWritable.class, redClass, + NullWritable.class, VectorWritable.class, SequenceFileOutputFormat.class, initialConf); + job.setCombinerClass(redClass); + job.setJobName("TimesSquaredJob: " + matrixInputPath); + + Configuration conf = job.getConfiguration(); + conf.set(INPUT_VECTOR, ivpURI.toString()); + conf.setBoolean(IS_SPARSE_OUTPUT, !v.isDense()); + conf.setInt(OUTPUT_VECTOR_DIMENSION, outputVectorDim); + + return job; + } + + public static Vector retrieveTimesSquaredOutputVector(Path outputVectorTmpPath, Configuration conf) + throws IOException { + Path outputFile = new Path(outputVectorTmpPath, OUTPUT_VECTOR_FILENAME + "/part-r-00000"); + SequenceFileValueIterator<VectorWritable> iterator = + new SequenceFileValueIterator<>(outputFile, true, conf); + try { + return iterator.next().get(); + } finally { + Closeables.close(iterator, true); + } + } + + public static class TimesSquaredMapper<T extends WritableComparable> + extends Mapper<T,VectorWritable, NullWritable,VectorWritable> { + + private Vector outputVector; + private Vector inputVector; + + Vector getOutputVector() { + return outputVector; + } + + @Override + protected void setup(Context ctx) throws IOException, InterruptedException { + try { + Configuration conf = ctx.getConfiguration(); + Path[] localFiles = DistributedCache.getLocalCacheFiles(conf); + Preconditions.checkArgument(localFiles != null && localFiles.length >= 1, + "missing paths from the DistributedCache"); + + Path inputVectorPath = HadoopUtil.getSingleCachedFile(conf); + + SequenceFileValueIterator<VectorWritable> iterator = + new SequenceFileValueIterator<>(inputVectorPath, true, conf); + try { + inputVector = iterator.next().get(); + } finally { + Closeables.close(iterator, true); + } + + int outDim = conf.getInt(OUTPUT_VECTOR_DIMENSION, Integer.MAX_VALUE); + outputVector = conf.getBoolean(IS_SPARSE_OUTPUT, false) + ? new RandomAccessSparseVector(outDim, 10) + : new DenseVector(outDim); + } catch (IOException ioe) { + throw new IllegalStateException(ioe); + } + } + + @Override + protected void map(T key, VectorWritable v, Context context) throws IOException, InterruptedException { + + double d = scale(v); + if (d == 1.0) { + outputVector.assign(v.get(), Functions.PLUS); + } else if (d != 0.0) { + outputVector.assign(v.get(), Functions.plusMult(d)); + } + } + + protected double scale(VectorWritable v) { + return v.get().dot(inputVector); + } + + @Override + protected void cleanup(Context ctx) throws IOException, InterruptedException { + ctx.write(NullWritable.get(), new VectorWritable(outputVector)); + } + + } + + public static class TimesMapper extends TimesSquaredMapper<IntWritable> { + + + @Override + protected void map(IntWritable rowNum, VectorWritable v, Context context) throws IOException, InterruptedException { + double d = scale(v); + if (d != 0.0) { + getOutputVector().setQuick(rowNum.get(), d); + } + } + } + + public static class VectorSummingReducer extends Reducer<NullWritable,VectorWritable,NullWritable,VectorWritable> { + + private Vector outputVector; + + @Override + protected void setup(Context ctx) throws IOException, InterruptedException { + Configuration conf = ctx.getConfiguration(); + int outputDimension = conf.getInt(OUTPUT_VECTOR_DIMENSION, Integer.MAX_VALUE); + outputVector = conf.getBoolean(IS_SPARSE_OUTPUT, false) + ? new RandomAccessSparseVector(outputDimension, 10) + : new DenseVector(outputDimension); + } + + @Override + protected void reduce(NullWritable key, Iterable<VectorWritable> vectors, Context ctx) + throws IOException, InterruptedException { + + for (VectorWritable v : vectors) { + if (v != null) { + outputVector.assign(v.get(), Functions.PLUS); + } + } + ctx.write(NullWritable.get(), new VectorWritable(outputVector)); + } + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java new file mode 100644 index 0000000..60066c6 --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java @@ -0,0 +1,85 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.hadoop; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; +import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; +import org.apache.hadoop.util.ToolRunner; +import org.apache.mahout.common.AbstractJob; +import org.apache.mahout.common.HadoopUtil; +import org.apache.mahout.common.mapreduce.MergeVectorsCombiner; +import org.apache.mahout.common.mapreduce.MergeVectorsReducer; +import org.apache.mahout.common.mapreduce.TransposeMapper; +import org.apache.mahout.math.VectorWritable; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** Transpose a matrix */ +public class TransposeJob extends AbstractJob { + + public static void main(String[] args) throws Exception { + ToolRunner.run(new TransposeJob(), args); + } + + @Override + public int run(String[] strings) throws Exception { + addInputOption(); + addOption("numRows", "nr", "Number of rows of the input matrix"); + addOption("numCols", "nc", "Number of columns of the input matrix"); + Map<String, List<String>> parsedArgs = parseArguments(strings); + if (parsedArgs == null) { + return -1; + } + + int numRows = Integer.parseInt(getOption("numRows")); + int numCols = Integer.parseInt(getOption("numCols")); + + DistributedRowMatrix matrix = new DistributedRowMatrix(getInputPath(), getTempPath(), numRows, numCols); + matrix.setConf(new Configuration(getConf())); + matrix.transpose(); + + return 0; + } + + public static Job buildTransposeJob(Path matrixInputPath, Path matrixOutputPath, int numInputRows) + throws IOException { + return buildTransposeJob(new Configuration(), matrixInputPath, matrixOutputPath, numInputRows); + } + + public static Job buildTransposeJob(Configuration initialConf, Path matrixInputPath, Path matrixOutputPath, + int numInputRows) throws IOException { + + Job job = HadoopUtil.prepareJob(matrixInputPath, matrixOutputPath, SequenceFileInputFormat.class, + TransposeMapper.class, IntWritable.class, VectorWritable.class, MergeVectorsReducer.class, IntWritable.class, + VectorWritable.class, SequenceFileOutputFormat.class, initialConf); + job.setCombinerClass(MergeVectorsCombiner.class); + job.getConfiguration().setInt(TransposeMapper.NEW_NUM_COLS_PARAM, numInputRows); + + job.setJobName("TransposeJob: " + matrixInputPath); + + return job; + } + + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/DistributedLanczosSolver.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/DistributedLanczosSolver.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/DistributedLanczosSolver.java new file mode 100644 index 0000000..86c0f6b --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/DistributedLanczosSolver.java @@ -0,0 +1,299 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.hadoop.decomposer; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +import com.google.common.io.Closeables; +import org.apache.hadoop.conf.Configurable; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.SequenceFile; +import org.apache.hadoop.io.Writable; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; +import org.apache.mahout.common.AbstractJob; +import org.apache.mahout.math.DenseVector; +import org.apache.mahout.math.NamedVector; +import org.apache.mahout.math.Vector; +import org.apache.mahout.math.VectorIterable; +import org.apache.mahout.math.VectorWritable; +import org.apache.mahout.math.decomposer.lanczos.LanczosSolver; +import org.apache.mahout.math.decomposer.lanczos.LanczosState; +import org.apache.mahout.math.hadoop.DistributedRowMatrix; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * See the SSVD code for a better option than using this: + * + * http://mahout.apache.org/users/dim-reduction/ssvd.html + * @see org.apache.mahout.math.hadoop.stochasticsvd.SSVDSolver + */ +@Deprecated +public class DistributedLanczosSolver extends LanczosSolver implements Tool { + + public static final String RAW_EIGENVECTORS = "rawEigenvectors"; + + private static final Logger log = LoggerFactory.getLogger(DistributedLanczosSolver.class); + + private Configuration conf; + + private Map<String, List<String>> parsedArgs; + + /** + * For the distributed case, the best guess at a useful initialization state for Lanczos we'll chose to be + * uniform over all input dimensions, L_2 normalized. + */ + public static Vector getInitialVector(VectorIterable corpus) { + Vector initialVector = new DenseVector(corpus.numCols()); + initialVector.assign(1.0 / Math.sqrt(corpus.numCols())); + return initialVector; + } + + public LanczosState runJob(Configuration originalConfig, + LanczosState state, + int desiredRank, + boolean isSymmetric, + String outputEigenVectorPathString) throws IOException { + ((Configurable) state.getCorpus()).setConf(new Configuration(originalConfig)); + setConf(originalConfig); + solve(state, desiredRank, isSymmetric); + serializeOutput(state, new Path(outputEigenVectorPathString)); + return state; + } + + /** + * Factored-out LanczosSolver for the purpose of invoking it programmatically + */ + public LanczosState runJob(Configuration originalConfig, + Path inputPath, + Path outputTmpPath, + int numRows, + int numCols, + boolean isSymmetric, + int desiredRank, + String outputEigenVectorPathString) throws IOException { + DistributedRowMatrix matrix = new DistributedRowMatrix(inputPath, outputTmpPath, numRows, numCols); + matrix.setConf(new Configuration(originalConfig)); + LanczosState state = new LanczosState(matrix, desiredRank, getInitialVector(matrix)); + return runJob(originalConfig, state, desiredRank, isSymmetric, outputEigenVectorPathString); + } + + @Override + public int run(String[] strings) throws Exception { + Path inputPath = new Path(AbstractJob.getOption(parsedArgs, "--input")); + Path outputPath = new Path(AbstractJob.getOption(parsedArgs, "--output")); + Path outputTmpPath = new Path(AbstractJob.getOption(parsedArgs, "--tempDir")); + Path workingDirPath = AbstractJob.getOption(parsedArgs, "--workingDir") != null + ? new Path(AbstractJob.getOption(parsedArgs, "--workingDir")) : null; + int numRows = Integer.parseInt(AbstractJob.getOption(parsedArgs, "--numRows")); + int numCols = Integer.parseInt(AbstractJob.getOption(parsedArgs, "--numCols")); + boolean isSymmetric = Boolean.parseBoolean(AbstractJob.getOption(parsedArgs, "--symmetric")); + int desiredRank = Integer.parseInt(AbstractJob.getOption(parsedArgs, "--rank")); + + boolean cleansvd = Boolean.parseBoolean(AbstractJob.getOption(parsedArgs, "--cleansvd")); + if (cleansvd) { + double maxError = Double.parseDouble(AbstractJob.getOption(parsedArgs, "--maxError")); + double minEigenvalue = Double.parseDouble(AbstractJob.getOption(parsedArgs, "--minEigenvalue")); + boolean inMemory = Boolean.parseBoolean(AbstractJob.getOption(parsedArgs, "--inMemory")); + return run(inputPath, + outputPath, + outputTmpPath, + workingDirPath, + numRows, + numCols, + isSymmetric, + desiredRank, + maxError, + minEigenvalue, + inMemory); + } + return run(inputPath, outputPath, outputTmpPath, workingDirPath, numRows, numCols, isSymmetric, desiredRank); + } + + /** + * Run the solver to produce raw eigenvectors, then run the EigenVerificationJob to clean them + * + * @param inputPath the Path to the input corpus + * @param outputPath the Path to the output + * @param outputTmpPath a Path to a temporary working directory + * @param numRows the int number of rows + * @param numCols the int number of columns + * @param isSymmetric true if the input matrix is symmetric + * @param desiredRank the int desired rank of eigenvectors to produce + * @param maxError the maximum allowable error + * @param minEigenvalue the minimum usable eigenvalue + * @param inMemory true if the verification can be done in memory + * @return an int indicating success (0) or otherwise + */ + public int run(Path inputPath, + Path outputPath, + Path outputTmpPath, + Path workingDirPath, + int numRows, + int numCols, + boolean isSymmetric, + int desiredRank, + double maxError, + double minEigenvalue, + boolean inMemory) throws Exception { + int result = run(inputPath, outputPath, outputTmpPath, workingDirPath, numRows, numCols, + isSymmetric, desiredRank); + if (result != 0) { + return result; + } + Path rawEigenVectorPath = new Path(outputPath, RAW_EIGENVECTORS); + return new EigenVerificationJob().run(inputPath, + rawEigenVectorPath, + outputPath, + outputTmpPath, + maxError, + minEigenvalue, + inMemory, + getConf() != null ? new Configuration(getConf()) : new Configuration()); + } + + /** + * Run the solver to produce the raw eigenvectors + * + * @param inputPath the Path to the input corpus + * @param outputPath the Path to the output + * @param outputTmpPath a Path to a temporary working directory + * @param numRows the int number of rows + * @param numCols the int number of columns + * @param isSymmetric true if the input matrix is symmetric + * @param desiredRank the int desired rank of eigenvectors to produce + * @return an int indicating success (0) or otherwise + */ + public int run(Path inputPath, + Path outputPath, + Path outputTmpPath, + Path workingDirPath, + int numRows, + int numCols, + boolean isSymmetric, + int desiredRank) throws Exception { + DistributedRowMatrix matrix = new DistributedRowMatrix(inputPath, outputTmpPath, numRows, numCols); + matrix.setConf(new Configuration(getConf() != null ? getConf() : new Configuration())); + + LanczosState state; + if (workingDirPath == null) { + state = new LanczosState(matrix, desiredRank, getInitialVector(matrix)); + } else { + HdfsBackedLanczosState hState = + new HdfsBackedLanczosState(matrix, desiredRank, getInitialVector(matrix), workingDirPath); + hState.setConf(matrix.getConf()); + state = hState; + } + solve(state, desiredRank, isSymmetric); + + Path outputEigenVectorPath = new Path(outputPath, RAW_EIGENVECTORS); + serializeOutput(state, outputEigenVectorPath); + return 0; + } + + /** + * @param state The final LanczosState to be serialized + * @param outputPath The path (relative to the current Configuration's FileSystem) to save the output to. + */ + public void serializeOutput(LanczosState state, Path outputPath) throws IOException { + int numEigenVectors = state.getIterationNumber(); + log.info("Persisting {} eigenVectors and eigenValues to: {}", numEigenVectors, outputPath); + Configuration conf = getConf() != null ? getConf() : new Configuration(); + FileSystem fs = FileSystem.get(outputPath.toUri(), conf); + SequenceFile.Writer seqWriter = + new SequenceFile.Writer(fs, conf, outputPath, IntWritable.class, VectorWritable.class); + try { + IntWritable iw = new IntWritable(); + for (int i = 0; i < numEigenVectors; i++) { + // Persist eigenvectors sorted by eigenvalues in descending order\ + NamedVector v = new NamedVector(state.getRightSingularVector(numEigenVectors - 1 - i), + "eigenVector" + i + ", eigenvalue = " + state.getSingularValue(numEigenVectors - 1 - i)); + Writable vw = new VectorWritable(v); + iw.set(i); + seqWriter.append(iw, vw); + } + } finally { + Closeables.close(seqWriter, false); + } + } + + @Override + public void setConf(Configuration configuration) { + conf = configuration; + } + + @Override + public Configuration getConf() { + return conf; + } + + public DistributedLanczosSolverJob job() { + return new DistributedLanczosSolverJob(); + } + + /** + * Inner subclass of AbstractJob so we get access to AbstractJob's functionality w.r.t. cmdline options, but still + * sublcass LanczosSolver. + */ + public class DistributedLanczosSolverJob extends AbstractJob { + @Override + public void setConf(Configuration conf) { + DistributedLanczosSolver.this.setConf(conf); + } + + @Override + public Configuration getConf() { + return DistributedLanczosSolver.this.getConf(); + } + + @Override + public int run(String[] args) throws Exception { + addInputOption(); + addOutputOption(); + addOption("numRows", "nr", "Number of rows of the input matrix"); + addOption("numCols", "nc", "Number of columns of the input matrix"); + addOption("rank", "r", "Desired decomposition rank (note: only roughly 1/4 to 1/3 " + + "of these will have the top portion of the spectrum)"); + addOption("symmetric", "sym", "Is the input matrix square and symmetric?"); + addOption("workingDir", "wd", "Working directory path to store Lanczos basis vectors " + + "(to be used on restarts, and to avoid too much RAM usage)"); + // options required to run cleansvd job + addOption("cleansvd", "cl", "Run the EigenVerificationJob to clean the eigenvectors after SVD", false); + addOption("maxError", "err", "Maximum acceptable error", "0.05"); + addOption("minEigenvalue", "mev", "Minimum eigenvalue to keep the vector for", "0.0"); + addOption("inMemory", "mem", "Buffer eigen matrix into memory (if you have enough!)", "false"); + + DistributedLanczosSolver.this.parsedArgs = parseArguments(args); + if (DistributedLanczosSolver.this.parsedArgs == null) { + return -1; + } else { + return DistributedLanczosSolver.this.run(args); + } + } + } + + public static void main(String[] args) throws Exception { + ToolRunner.run(new DistributedLanczosSolver().job(), args); + } +} http://git-wip-us.apache.org/repos/asf/mahout/blob/5eda9e1f/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/EigenVector.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/EigenVector.java b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/EigenVector.java new file mode 100644 index 0000000..d2f0c8c --- /dev/null +++ b/community/mahout-mr/src/main/java/org/apache/mahout/math/hadoop/decomposer/EigenVector.java @@ -0,0 +1,76 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.mahout.math.hadoop.decomposer; + +import org.apache.mahout.math.DenseVector; +import org.apache.mahout.math.NamedVector; +import org.apache.mahout.math.Vector; + +import java.util.regex.Pattern; + +/** + * TODO this is a horrible hack. Make a proper writable subclass also. + */ +public class EigenVector extends NamedVector { + + private static final Pattern EQUAL_PATTERN = Pattern.compile(" = "); + private static final Pattern PIPE_PATTERN = Pattern.compile("\\|"); + + public EigenVector(Vector v, double eigenValue, double cosAngleError, int order) { + super(v instanceof DenseVector ? (DenseVector) v : new DenseVector(v), + "e|" + order + "| = |" + eigenValue + "|, err = " + cosAngleError); + } + + public double getEigenValue() { + return getEigenValue(getName()); + } + + public double getCosAngleError() { + return getCosAngleError(getName()); + } + + public int getIndex() { + return getIndex(getName()); + } + + public static double getEigenValue(CharSequence name) { + return parseMetaData(name)[1]; + } + + public static double getCosAngleError(CharSequence name) { + return parseMetaData(name)[2]; + } + + public static int getIndex(CharSequence name) { + return (int)parseMetaData(name)[0]; + } + + public static double[] parseMetaData(CharSequence name) { + double[] m = new double[3]; + String[] s = EQUAL_PATTERN.split(name); + m[0] = Double.parseDouble(PIPE_PATTERN.split(s[0])[1]); + m[1] = Double.parseDouble(PIPE_PATTERN.split(s[1])[1]); + m[2] = Double.parseDouble(s[2].substring(1)); + return m; + } + + protected double[] parseMetaData() { + return parseMetaData(getName()); + } + +}
