You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by sr...@apache.org on 2011/03/31 12:40:07 UTC
svn commit: r1087245 - in /mahout/trunk/core/src:
main/java/org/apache/mahout/math/hadoop/
test/java/org/apache/mahout/math/hadoop/
Author: srowen
Date: Thu Mar 31 10:40:07 2011
New Revision: 1087245
URL: http://svn.apache.org/viewvc?rev=1087245&view=rev
Log:
MAHOUT-641 Pass Configuration into and through distributed matrix operations correctly
Modified:
mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java
mahout/trunk/core/src/test/java/org/apache/mahout/math/hadoop/TestDistributedRowMatrix.java
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java?rev=1087245&r1=1087244&r2=1087245&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/DistributedRowMatrix.java Thu Mar 31 10:40:07 2011
@@ -149,8 +149,14 @@ public class DistributedRowMatrix implem
throw new CardinalityException(numRows, other.numRows());
}
Path outPath = new Path(outputTmpBasePath.getParent(), "productWith-" + (System.nanoTime() & 0xFF));
+
+ Configuration initialConf = getConf() == null ? new Configuration() : getConf();
Configuration conf =
- MatrixMultiplicationJob.createMatrixMultiplyJobConf(rowPath, other.rowPath, outPath, other.numCols);
+ 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);
@@ -159,7 +165,8 @@ public class DistributedRowMatrix implem
public DistributedRowMatrix transpose() throws IOException {
Path outputPath = new Path(rowPath.getParent(), "transpose-" + (System.nanoTime() & 0xFF));
- Configuration conf = TransposeJob.buildTransposeJobConf(rowPath, outputPath, numRows);
+ Configuration initialConf = getConf() == null ? new Configuration() : getConf();
+ Configuration conf = TransposeJob.buildTransposeJobConf(initialConf, rowPath, outputPath, numRows);
JobClient.runJob(new JobConf(conf));
DistributedRowMatrix m = new DistributedRowMatrix(outputPath, outputTmpPath, numCols, numRows);
m.setConf(this.conf);
@@ -169,8 +176,10 @@ public class DistributedRowMatrix implem
@Override
public Vector times(Vector v) {
try {
+ Configuration initialConf = getConf() == null ? new Configuration() : getConf();
Configuration conf =
- TimesSquaredJob.createTimesJobConf(v,
+ TimesSquaredJob.createTimesJobConf(initialConf,
+ v,
numRows,
rowPath,
new Path(outputTmpPath, Long.toString(System.nanoTime())));
@@ -184,8 +193,10 @@ public class DistributedRowMatrix implem
@Override
public Vector timesSquared(Vector v) {
try {
+ Configuration initialConf = getConf() == null ? new Configuration() : getConf();
Configuration conf =
- TimesSquaredJob.createTimesSquaredJobConf(v,
+ TimesSquaredJob.createTimesSquaredJobConf(initialConf,
+ v,
rowPath,
new Path(outputTmpBasePath,
new Path(Long.toString(System.nanoTime()))));
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java?rev=1087245&r1=1087244&r2=1087245&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/MatrixMultiplicationJob.java Thu Mar 31 10:40:07 2011
@@ -46,8 +46,19 @@ public class MatrixMultiplicationJob ext
private static final String OUT_CARD = "output.vector.cardinality";
- public static Configuration createMatrixMultiplyJobConf(Path aPath, Path bPath, Path outPath, int outCardinality) {
- JobConf conf = new JobConf(MatrixMultiplicationJob.class);
+ 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));
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java?rev=1087245&r1=1087244&r2=1087245&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TimesSquaredJob.java Thu Mar 31 10:40:07 2011
@@ -60,10 +60,17 @@ public final class TimesSquaredJob {
private TimesSquaredJob() { }
- public static Configuration createTimesSquaredJobConf(Vector v,
+ public static Configuration createTimesSquaredJobConf(Vector v, Path matrixInputPath, Path outputVectorPath)
+ throws IOException {
+ return createTimesSquaredJobConf(new Configuration(), v, matrixInputPath, outputVectorPath);
+ }
+
+ public static Configuration createTimesSquaredJobConf(Configuration initialConf,
+ Vector v,
Path matrixInputPath,
Path outputVectorPath) throws IOException {
- return createTimesSquaredJobConf(v,
+ return createTimesSquaredJobConf(initialConf,
+ v,
matrixInputPath,
outputVectorPath,
TimesSquaredMapper.class,
@@ -74,7 +81,16 @@ public final class TimesSquaredJob {
int outDim,
Path matrixInputPath,
Path outputVectorPath) throws IOException {
- return createTimesSquaredJobConf(v,
+ return createTimesJobConf(new Configuration(), v, outDim, matrixInputPath, outputVectorPath);
+ }
+
+ public static Configuration createTimesJobConf(Configuration initialConf,
+ Vector v,
+ int outDim,
+ Path matrixInputPath,
+ Path outputVectorPath) throws IOException {
+ return createTimesSquaredJobConf(initialConf,
+ v,
outDim,
matrixInputPath,
outputVectorPath,
@@ -82,14 +98,29 @@ public final class TimesSquaredJob {
VectorSummingReducer.class);
}
-
public static Configuration createTimesSquaredJobConf(Vector v,
Path matrixInputPath,
Path outputVectorPathBase,
Class<? extends TimesSquaredMapper> mapClass,
Class<? extends VectorSummingReducer> redClass)
throws IOException {
- return createTimesSquaredJobConf(v, v.size(), matrixInputPath, outputVectorPathBase, mapClass, redClass);
+ return createTimesSquaredJobConf(new Configuration(), v, matrixInputPath, outputVectorPathBase, mapClass, redClass);
+ }
+
+ public static Configuration createTimesSquaredJobConf(Configuration initialConf,
+ Vector v,
+ Path matrixInputPath,
+ Path outputVectorPathBase,
+ Class<? extends TimesSquaredMapper> mapClass,
+ Class<? extends VectorSummingReducer> redClass)
+ throws IOException {
+ return createTimesSquaredJobConf(initialConf,
+ v,
+ v.size(),
+ matrixInputPath,
+ outputVectorPathBase,
+ mapClass,
+ redClass);
}
public static Configuration createTimesSquaredJobConf(Vector v,
@@ -99,7 +130,25 @@ public final class TimesSquaredJob {
Class<? extends TimesSquaredMapper> mapClass,
Class<? extends VectorSummingReducer> redClass)
throws IOException {
- JobConf conf = new JobConf(TimesSquaredJob.class);
+
+ return createTimesSquaredJobConf(new Configuration(),
+ v,
+ outputVectorDim,
+ matrixInputPath,
+ outputVectorPathBase,
+ mapClass,
+ redClass);
+ }
+
+ public static Configuration createTimesSquaredJobConf(Configuration initialConf,
+ Vector v,
+ int outputVectorDim,
+ Path matrixInputPath,
+ Path outputVectorPathBase,
+ Class<? extends TimesSquaredMapper> mapClass,
+ Class<? extends VectorSummingReducer> redClass)
+ throws IOException {
+ JobConf conf = new JobConf(initialConf, TimesSquaredJob.class);
conf.setJobName("TimesSquaredJob: " + matrixInputPath);
FileSystem fs = FileSystem.get(conf);
matrixInputPath = fs.makeQualified(matrixInputPath);
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java?rev=1087245&r1=1087244&r2=1087245&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/hadoop/TransposeJob.java Thu Mar 31 10:40:07 2011
@@ -78,7 +78,14 @@ public class TransposeJob extends Abstra
public static Configuration buildTransposeJobConf(Path matrixInputPath,
Path matrixOutputPath,
int numInputRows) throws IOException {
- JobConf conf = new JobConf(TransposeJob.class);
+ return buildTransposeJobConf(new Configuration(), matrixInputPath, matrixOutputPath, numInputRows);
+ }
+
+ public static Configuration buildTransposeJobConf(Configuration initialConf,
+ Path matrixInputPath,
+ Path matrixOutputPath,
+ int numInputRows) throws IOException {
+ JobConf conf = new JobConf(initialConf, TransposeJob.class);
conf.setJobName("TransposeJob: " + matrixInputPath + " transpose -> " + matrixOutputPath);
FileSystem fs = FileSystem.get(conf);
matrixInputPath = fs.makeQualified(matrixInputPath);
Modified: mahout/trunk/core/src/test/java/org/apache/mahout/math/hadoop/TestDistributedRowMatrix.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/test/java/org/apache/mahout/math/hadoop/TestDistributedRowMatrix.java?rev=1087245&r1=1087244&r2=1087245&view=diff
==============================================================================
--- mahout/trunk/core/src/test/java/org/apache/mahout/math/hadoop/TestDistributedRowMatrix.java (original)
+++ mahout/trunk/core/src/test/java/org/apache/mahout/math/hadoop/TestDistributedRowMatrix.java Thu Mar 31 10:40:07 2011
@@ -17,6 +17,11 @@
package org.apache.mahout.math.hadoop;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -31,12 +36,9 @@ import org.apache.mahout.math.VectorWrit
import org.apache.mahout.math.decomposer.SolverTest;
import org.junit.Test;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
public final class TestDistributedRowMatrix extends MahoutTestCase {
+ public static final String TEST_PROPERTY_KEY = "test.property.key";
+ public static final String TEST_PROPERTY_VALUE = "test.property.value";
private static void assertEquals(VectorIterable m, VectorIterable mtt, double errorTolerance) {
Iterator<MatrixSlice> mIt = m.iterateAll();
@@ -108,6 +110,102 @@ public final class TestDistributedRowMat
assertEquals(expected, product, EPSILON);
}
+ @Test
+ public void testMatrixMultiplactionJobConfBuilder() throws Exception {
+ Configuration initialConf = createInitialConf();
+
+ Path baseTmpDirPath = getTestTempDirPath("testpaths");
+ Path aPath = new Path(baseTmpDirPath, "a");
+ Path bPath = new Path(baseTmpDirPath, "b");
+ Path outPath = new Path(baseTmpDirPath, "out");
+
+ Configuration mmJobConf = MatrixMultiplicationJob.createMatrixMultiplyJobConf(aPath, bPath, outPath, 10);
+ Configuration mmCustomJobConf = MatrixMultiplicationJob.createMatrixMultiplyJobConf(initialConf,
+ aPath,
+ bPath,
+ outPath,
+ 10);
+
+ assertNull(mmJobConf.get(TEST_PROPERTY_KEY));
+ assertEquals(TEST_PROPERTY_VALUE, mmCustomJobConf.get(TEST_PROPERTY_KEY));
+ }
+
+ @Test
+ public void testTransposeJobConfBuilder() throws Exception {
+ Configuration initialConf = createInitialConf();
+
+ Path baseTmpDirPath = getTestTempDirPath("testpaths");
+ Path inputPath = new Path(baseTmpDirPath, "input");
+ Path outputPath = new Path(baseTmpDirPath, "output");
+
+ Configuration transposeJobConf = TransposeJob.buildTransposeJobConf(inputPath, outputPath, 10);
+ Configuration transposeCustomJobConf = TransposeJob.buildTransposeJobConf(initialConf, inputPath, outputPath, 10);
+
+ assertNull(transposeJobConf.get(TEST_PROPERTY_KEY));
+ assertEquals(TEST_PROPERTY_VALUE, transposeCustomJobConf.get(TEST_PROPERTY_KEY));
+ }
+
+ @Test public void testTimesSquaredJobConfBuilders() throws Exception {
+ Configuration initialConf = createInitialConf();
+
+ Path baseTmpDirPath = getTestTempDirPath("testpaths");
+ Path inputPath = new Path(baseTmpDirPath, "input");
+ Path outputPath = new Path(baseTmpDirPath, "output");
+
+ Vector v = new RandomAccessSparseVector(50);
+ v.assign(1.0);
+
+ Configuration timesSquaredJobConf1 = TimesSquaredJob.createTimesSquaredJobConf(v, inputPath, outputPath);
+ Configuration customTimesSquaredJobConf1 = TimesSquaredJob.createTimesSquaredJobConf(initialConf, v, inputPath, outputPath);
+
+ assertNull(timesSquaredJobConf1.get(TEST_PROPERTY_KEY));
+ assertEquals(TEST_PROPERTY_VALUE, customTimesSquaredJobConf1.get(TEST_PROPERTY_KEY));
+
+ Configuration timesJobConf = TimesSquaredJob.createTimesJobConf(v, 50, inputPath, outputPath);
+ Configuration customTimesJobConf = TimesSquaredJob.createTimesJobConf(initialConf, v, 50, inputPath, outputPath);
+
+ assertNull(timesJobConf.get(TEST_PROPERTY_KEY));
+ assertEquals(TEST_PROPERTY_VALUE, customTimesJobConf.get(TEST_PROPERTY_KEY));
+
+ Configuration timesSquaredJobConf2 = TimesSquaredJob.createTimesSquaredJobConf(v,
+ inputPath,
+ outputPath,
+ TimesSquaredJob.TimesSquaredMapper.class,
+ TimesSquaredJob.VectorSummingReducer.class);
+ Configuration customTimesSquaredJobConf2 = TimesSquaredJob.createTimesSquaredJobConf(initialConf,
+ v,
+ inputPath,
+ outputPath,
+ TimesSquaredJob.TimesSquaredMapper.class,
+ TimesSquaredJob.VectorSummingReducer.class);
+
+ assertNull(timesSquaredJobConf2.get(TEST_PROPERTY_KEY));
+ assertEquals(TEST_PROPERTY_VALUE, customTimesSquaredJobConf2.get(TEST_PROPERTY_KEY));
+
+ Configuration timesSquaredJobConf3 = TimesSquaredJob.createTimesSquaredJobConf(v,
+ 50,
+ inputPath,
+ outputPath,
+ TimesSquaredJob.TimesSquaredMapper.class,
+ TimesSquaredJob.VectorSummingReducer.class);
+ Configuration customTimesSquaredJobConf3 = TimesSquaredJob.createTimesSquaredJobConf(initialConf,
+ v,
+ 50,
+ inputPath,
+ outputPath,
+ TimesSquaredJob.TimesSquaredMapper.class,
+ TimesSquaredJob.VectorSummingReducer.class);
+
+ assertNull(timesSquaredJobConf3.get(TEST_PROPERTY_KEY));
+ assertEquals(TEST_PROPERTY_VALUE, customTimesSquaredJobConf3.get(TEST_PROPERTY_KEY));
+ }
+
+ public Configuration createInitialConf() {
+ Configuration initialConf = new Configuration();
+ initialConf.set(TEST_PROPERTY_KEY, TEST_PROPERTY_VALUE);
+ return initialConf;
+ }
+
public DistributedRowMatrix randomDistributedMatrix(int numRows,
int nonNullRows,
int numCols,