You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by td...@apache.org on 2010/12/30 03:30:04 UTC
svn commit: r1053787 [1/2] - in /mahout/trunk:
core/src/main/java/org/apache/mahout/classifier/sgd/
core/src/main/java/org/apache/mahout/ep/
core/src/main/java/org/apache/mahout/math/
core/src/main/java/org/apache/mahout/math/stats/ core/src/test/java/...
Author: tdunning
Date: Thu Dec 30 02:30:03 2010
New Revision: 1053787
URL: http://svn.apache.org/viewvc?rev=1053787&view=rev
Log:
MAHOUT-545 - Big patch to support writable classifiers.
Added:
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PolymorphicWritable.java
Modified:
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AbstractOnlineLogisticRegression.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AdaptiveLogisticRegression.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/CrossFoldLearner.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ElasticBandPrior.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L1.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L2.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ModelSerializer.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/OnlineLogisticRegression.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PriorFunction.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/TPrior.java
mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/UniformPrior.java
mahout/trunk/core/src/main/java/org/apache/mahout/ep/EvolutionaryProcess.java
mahout/trunk/core/src/main/java/org/apache/mahout/ep/Mapping.java
mahout/trunk/core/src/main/java/org/apache/mahout/ep/Payload.java
mahout/trunk/core/src/main/java/org/apache/mahout/ep/State.java
mahout/trunk/core/src/main/java/org/apache/mahout/ep/ThreadedEvolutionaryProcess.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/MatrixWritable.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/VectorWritable.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GlobalOnlineAuc.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GroupedOnlineAuc.java
mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/OnlineAuc.java
mahout/trunk/core/src/test/java/org/apache/mahout/classifier/sgd/ModelSerializerTest.java
mahout/trunk/core/src/test/java/org/apache/mahout/ep/EvolutionaryProcessTest.java
mahout/trunk/examples/src/main/java/org/apache/mahout/classifier/sgd/TrainNewsGroups.java
mahout/trunk/math/src/main/java/org/apache/mahout/math/SparseMatrix.java
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AbstractOnlineLogisticRegression.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AbstractOnlineLogisticRegression.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AbstractOnlineLogisticRegression.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AbstractOnlineLogisticRegression.java Thu Dec 30 02:30:03 2010
@@ -17,6 +17,7 @@
package org.apache.mahout.classifier.sgd;
+import com.google.common.base.Preconditions;
import org.apache.mahout.classifier.AbstractVectorClassifier;
import org.apache.mahout.classifier.OnlineLearner;
import org.apache.mahout.math.DenseVector;
@@ -25,8 +26,6 @@ import org.apache.mahout.math.Vector;
import org.apache.mahout.math.function.Functions;
import org.apache.mahout.math.function.UnaryFunction;
-import com.google.common.base.Preconditions;
-
import java.util.Iterator;
/**
@@ -46,7 +45,7 @@ public abstract class AbstractOnlineLogi
// number of categories we are classifying. This should the number of rows of beta plus one.
protected int numCategories;
- private int step = 0;
+ protected int step = 0;
// information about how long since coefficient rows were updated. This allows lazy regularization.
protected Vector updateSteps;
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AdaptiveLogisticRegression.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AdaptiveLogisticRegression.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AdaptiveLogisticRegression.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/AdaptiveLogisticRegression.java Thu Dec 30 02:30:03 2010
@@ -18,14 +18,19 @@
package org.apache.mahout.classifier.sgd;
import com.google.common.collect.Lists;
+import org.apache.hadoop.io.Writable;
import org.apache.mahout.classifier.OnlineLearner;
import org.apache.mahout.ep.Payload;
import org.apache.mahout.ep.EvolutionaryProcess;
import org.apache.mahout.ep.Mapping;
import org.apache.mahout.ep.State;
import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.VectorWritable;
import org.apache.mahout.math.stats.OnlineAuc;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.ExecutionException;
@@ -46,19 +51,19 @@ import java.util.concurrent.ExecutionExc
* increase the cost of training since if we are using 5-fold cross-validation, each vector is used
* 4 times for training and once for classification. If this becomes a problem, then we should
* probably use a 2-way unbalanced train/test split rather than full cross validation. With the
- * current default settings, we have 100 learners running. This is better than the alternative
- * of running hundreds of training passes to find good hyper-parameters because we only have to
- * parse and feature-ize our inputs once. If you already have good hyper-parameters, then you
- * might prefer to just run one CrossFoldLearner with those settings.
+ * current default settings, we have 100 learners running. This is better than the alternative of
+ * running hundreds of training passes to find good hyper-parameters because we only have to parse
+ * and feature-ize our inputs once. If you already have good hyper-parameters, then you might
+ * prefer to just run one CrossFoldLearner with those settings.
* <p/>
- * The fitness used here is AUC. Another alternative would be to try log-likelihood, but it is
- * much easier to get bogus values of log-likelihood than with AUC and the results seem to
- * accord pretty well. It would be nice to allow the fitness function to be pluggable. This
- * use of AUC means that AdaptiveLogisticRegression is mostly suited for binary target variables.
- * This will be fixed before long by extending OnlineAuc to handle non-binary cases or by using
- * a different fitness value in non-binary cases.
+ * The fitness used here is AUC. Another alternative would be to try log-likelihood, but it is much
+ * easier to get bogus values of log-likelihood than with AUC and the results seem to accord pretty
+ * well. It would be nice to allow the fitness function to be pluggable. This use of AUC means that
+ * AdaptiveLogisticRegression is mostly suited for binary target variables. This will be fixed
+ * before long by extending OnlineAuc to handle non-binary cases or by using a different fitness
+ * value in non-binary cases.
*/
-public class AdaptiveLogisticRegression implements OnlineLearner {
+public class AdaptiveLogisticRegression implements OnlineLearner, Writable {
private static final int SURVIVORS = 2;
@@ -70,23 +75,24 @@ public class AdaptiveLogisticRegression
private int bufferSize = 1000;
private List<TrainingExample> buffer = Lists.newArrayList();
- private EvolutionaryProcess<Wrapper> ep;
- private State<Wrapper> best;
+ private EvolutionaryProcess<Wrapper, CrossFoldLearner> ep;
+ private State<Wrapper, CrossFoldLearner> best;
private int threadCount = 20;
private int poolSize = 20;
- private State<Wrapper> seed;
+ private State<Wrapper, CrossFoldLearner> seed;
private int numFeatures;
private boolean freezeSurvivors = true;
- // for GSON
- private AdaptiveLogisticRegression() {
+ public AdaptiveLogisticRegression() {
}
public AdaptiveLogisticRegression(int numCategories, int numFeatures, PriorFunction prior) {
this.numFeatures = numFeatures;
- seed = new State<Wrapper>(new double[2], 10);
+ seed = new State<Wrapper, CrossFoldLearner>(new double[2], 10);
Wrapper w = new Wrapper(numCategories, numFeatures, prior);
+ seed.setPayload(w);
+
w.setMappings(seed);
seed.setPayload(w);
setPoolSize(poolSize);
@@ -114,9 +120,10 @@ public class AdaptiveLogisticRegression
private void trainWithBufferedExamples() {
try {
- this.best = ep.parallelDo(new EvolutionaryProcess.Function<Wrapper>() {
+ this.best = ep.parallelDo(new EvolutionaryProcess.Function<Payload<CrossFoldLearner>>() {
@Override
- public double apply(Wrapper x, double[] params) {
+ public double apply(Payload<CrossFoldLearner> z, double[] params) {
+ Wrapper x = (Wrapper) z;
for (TrainingExample example : buffer) {
x.train(example);
}
@@ -148,7 +155,7 @@ public class AdaptiveLogisticRegression
// now grossly hack the top survivors so they stick around. Set their
// mutation rates small and also hack their learning rate to be small
// as well.
- for (State<Wrapper> state : ep.getPopulation().subList(0, SURVIVORS)) {
+ for (State<Wrapper, CrossFoldLearner> state : ep.getPopulation().subList(0, SURVIVORS)) {
state.getPayload().freeze(state);
}
}
@@ -188,11 +195,12 @@ public class AdaptiveLogisticRegression
public void close() {
trainWithBufferedExamples();
try {
- ep.parallelDo(new EvolutionaryProcess.Function<Wrapper>() {
+ ep.parallelDo(new EvolutionaryProcess.Function<Payload<CrossFoldLearner>>() {
@Override
- public double apply(Wrapper payload, double[] params) {
- payload.getLearner().close();
- return payload.getLearner().logLikelihood();
+ public double apply(Payload<CrossFoldLearner> payload, double[] params) {
+ CrossFoldLearner learner = ((Wrapper) payload).getLearner();
+ learner.close();
+ return learner.logLikelihood();
}
});
ep.close();
@@ -205,7 +213,8 @@ public class AdaptiveLogisticRegression
/**
* How often should the evolutionary optimization of learning parameters occur?
- * @param interval Number of training examples to use in each epoch of optimization.
+ *
+ * @param interval Number of training examples to use in each epoch of optimization.
*/
public void setInterval(int interval) {
setInterval(interval, interval);
@@ -216,8 +225,8 @@ public class AdaptiveLogisticRegression
* number of steps per decade. Note that values < 200 are not accepted. Values even that small
* are unlikely to be useful.
*
- * @param minInterval The minimum epoch length for the evolutionary optimization
- * @param maxInterval The maximum epoch length
+ * @param minInterval The minimum epoch length for the evolutionary optimization
+ * @param maxInterval The maximum epoch length
*/
public void setInterval(int minInterval, int maxInterval) {
this.minInterval = Math.max(200, minInterval);
@@ -243,13 +252,13 @@ public class AdaptiveLogisticRegression
}
private void setupOptimizer(int poolSize) {
- ep = new EvolutionaryProcess<Wrapper>(threadCount, poolSize, seed);
+ ep = new EvolutionaryProcess<Wrapper, CrossFoldLearner>(threadCount, poolSize, seed);
}
/**
- * Returns the size of the internal feature vector. Note that this is not the
- * same as the number of distinct features, especially if feature hashing is
- * being used.
+ * Returns the size of the internal feature vector. Note that this is not the same as the number
+ * of distinct features, especially if feature hashing is being used.
+ *
* @return The internal feature vector size.
*/
public int numFeatures() {
@@ -257,23 +266,25 @@ public class AdaptiveLogisticRegression
}
/**
- * What is the AUC for the current best member of the population. If no member is best,
- * usually because we haven't done any training yet, then the result is set to NaN.
- * @return The AUC of the best member of the population or NaN if we can't figure that out.
+ * What is the AUC for the current best member of the population. If no member is best, usually
+ * because we haven't done any training yet, then the result is set to NaN.
+ *
+ * @return The AUC of the best member of the population or NaN if we can't figure that out.
*/
public double auc() {
if (best == null) {
return Double.NaN;
} else {
- return best.getPayload().getLearner().auc();
+ Wrapper payload = best.getPayload();
+ return payload.getLearner().auc();
}
}
- public State<Wrapper> getBest() {
+ public State<Wrapper, CrossFoldLearner> getBest() {
return best;
}
- public void setBest(State<Wrapper> best) {
+ public void setBest(State<Wrapper, CrossFoldLearner> best) {
this.best = best;
}
@@ -309,19 +320,19 @@ public class AdaptiveLogisticRegression
return buffer;
}
- public EvolutionaryProcess<Wrapper> getEp() {
+ public EvolutionaryProcess<Wrapper, CrossFoldLearner> getEp() {
return ep;
}
- public void setEp(EvolutionaryProcess<Wrapper> ep) {
+ public void setEp(EvolutionaryProcess<Wrapper, CrossFoldLearner> ep) {
this.ep = ep;
}
- public State<Wrapper> getSeed() {
+ public State<Wrapper, CrossFoldLearner> getSeed() {
return seed;
}
- public void setSeed(State<Wrapper> seed) {
+ public void setSeed(State<Wrapper, CrossFoldLearner> seed) {
this.seed = seed;
}
@@ -344,15 +355,14 @@ public class AdaptiveLogisticRegression
* params in the following order <ul> <li> regularization constant lambda <li> learningRate </ul>.
* All other parameters are set in such a way so as to defeat annealing to the extent possible.
* This lets the evolutionary algorithm handle the annealing.
- * <p>
+ * <p/>
* Note that per coefficient annealing is still done and no optimization of the per coefficient
* offset is done.
*/
- public static class Wrapper implements Payload<Wrapper> {
+ public static class Wrapper implements Payload<CrossFoldLearner> {
private CrossFoldLearner wrapped;
- private Wrapper() {
- // just here to help copy
+ public Wrapper() {
}
public Wrapper(int numCategories, int numFeatures, PriorFunction prior) {
@@ -377,7 +387,7 @@ public class AdaptiveLogisticRegression
wrapped.decayExponent(0);
}
- public void freeze(State<Wrapper> s) {
+ public void freeze(State<Wrapper, CrossFoldLearner> s) {
// radically decrease learning rate
s.getParams()[1] -= 10;
@@ -389,7 +399,7 @@ public class AdaptiveLogisticRegression
}
}
- public void setMappings(State<Wrapper> x) {
+ public void setMappings(State<Wrapper, CrossFoldLearner> x) {
int i = 0;
// set the range for regularization (lambda)
x.setMap(i++, Mapping.logLimit(1.0e-8, 0.1));
@@ -413,15 +423,27 @@ public class AdaptiveLogisticRegression
public void setAucEvaluator(OnlineAuc auc) {
wrapped.setAucEvaluator(auc);
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ wrapped.write(out);
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ wrapped = new CrossFoldLearner();
+ wrapped.readFields(input);
+ }
}
- public static class TrainingExample {
+ public static class TrainingExample implements Writable {
private long key;
private String groupKey;
private int actual;
private Vector instance;
// for GSON
+
private TrainingExample() {
}
@@ -447,6 +469,87 @@ public class AdaptiveLogisticRegression
public String getGroupKey() {
return groupKey;
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeLong(key);
+ if (groupKey != null) {
+ out.writeBoolean(true);
+ out.writeUTF(groupKey);
+ } else {
+ out.writeBoolean(false);
+ }
+ out.writeInt(actual);
+ VectorWritable.writeVector(out, instance, true);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ key = in.readLong();
+ if (in.readBoolean()) {
+ groupKey = in.readUTF();
+ }
+ actual = in.readInt();
+ instance = VectorWritable.readVector(in);
+ }
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(record);
+ out.writeInt(cutoff);
+ out.writeInt(minInterval);
+ out.writeInt(maxInterval);
+ out.writeInt(currentStep);
+ out.writeInt(bufferSize);
+
+ out.writeInt(buffer.size());
+ for (TrainingExample example : buffer) {
+ example.write(out);
+ }
+
+ ep.write(out);
+
+ best.write(out);
+
+ out.writeInt(threadCount);
+ out.writeInt(poolSize);
+ seed.write(out);
+ out.writeInt(numFeatures);
+
+ out.writeBoolean(freezeSurvivors);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ record = in.readInt();
+ cutoff = in.readInt();
+ minInterval = in.readInt();
+ maxInterval = in.readInt();
+ currentStep = in.readInt();
+ bufferSize = in.readInt();
+
+ int n = in.readInt();
+ buffer = Lists.newArrayList();
+ for (int i = 0; i < n; i++) {
+ TrainingExample example = new TrainingExample();
+ example.readFields(in);
+ buffer.add(example);
+ }
+
+ ep = new EvolutionaryProcess<Wrapper, CrossFoldLearner>();
+ ep.readFields(in);
+
+ best = new State<Wrapper, CrossFoldLearner>();
+ best.readFields(in);
+
+ threadCount = in.readInt();
+ poolSize = in.readInt();
+ seed = new State<Wrapper, CrossFoldLearner>();
+ seed.readFields(in);
+
+ numFeatures = in.readInt();
+ freezeSurvivors = in.readBoolean();
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/CrossFoldLearner.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/CrossFoldLearner.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/CrossFoldLearner.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/CrossFoldLearner.java Thu Dec 30 02:30:03 2010
@@ -18,15 +18,19 @@
package org.apache.mahout.classifier.sgd;
import com.google.common.collect.Lists;
+import org.apache.hadoop.io.Writable;
import org.apache.mahout.classifier.AbstractVectorClassifier;
import org.apache.mahout.classifier.OnlineLearner;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.function.BinaryFunction;
import org.apache.mahout.math.function.Functions;
-import org.apache.mahout.math.stats.OnlineAuc;
import org.apache.mahout.math.stats.GlobalOnlineAuc;
+import org.apache.mahout.math.stats.OnlineAuc;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.List;
/**
@@ -37,7 +41,7 @@ import java.util.List;
* time the training data is traversed or a tracking key such as the file offset of the training
* record should be passed with each training example.
*/
-public class CrossFoldLearner extends AbstractVectorClassifier implements OnlineLearner {
+public class CrossFoldLearner extends AbstractVectorClassifier implements OnlineLearner, Writable {
private int record;
// minimum score to be used for computing log likelihood
private static final double MIN_SCORE = 1.0e-50;
@@ -282,4 +286,44 @@ public class CrossFoldLearner extends Ab
public void setPrior(PriorFunction prior) {
this.prior = prior;
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(record);
+ PolymorphicWritable.write(out, auc);
+ out.writeDouble(logLikelihood);
+ out.writeInt(models.size());
+ for (OnlineLogisticRegression model : models) {
+ model.write(out);
+ }
+
+ for (double x : parameters) {
+ out.writeDouble(x);
+ }
+ out.writeInt(numFeatures);
+ PolymorphicWritable.write(out, prior);
+ out.writeDouble(percentCorrect);
+ out.writeInt(windowSize);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ record = in.readInt();
+ auc = PolymorphicWritable.read(in, OnlineAuc.class);
+ logLikelihood = in.readDouble();
+ int n = in.readInt();
+ for (int i = 0; i < n; i++) {
+ OnlineLogisticRegression olr = new OnlineLogisticRegression();
+ olr.readFields(in);
+ models.add(olr);
+ }
+ parameters = new double[4];
+ for (int i = 0; i < 4; i++) {
+ parameters[i] = in.readDouble();
+ }
+ numFeatures = in.readInt();
+ prior = PolymorphicWritable.read(in, PriorFunction.class);
+ percentCorrect = in.readDouble();
+ windowSize = in.readInt();
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ElasticBandPrior.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ElasticBandPrior.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ElasticBandPrior.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ElasticBandPrior.java Thu Dec 30 02:30:03 2010
@@ -17,14 +17,18 @@
package org.apache.mahout.classifier.sgd;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
/**
* Implements a linear combination of L1 and L2 priors. This can give an
* interesting mixture of sparsity and load-sharing between redundant predictors.
*/
public class ElasticBandPrior implements PriorFunction {
- private final double alphaByLambda;
- private final L1 l1;
- private final L2 l2;
+ private double alphaByLambda;
+ private L1 l1;
+ private L2 l2;
public ElasticBandPrior(double alphaByLambda) {
this.alphaByLambda = alphaByLambda;
@@ -48,4 +52,20 @@ public class ElasticBandPrior implements
public double logP(double betaIJ) {
return l1.logP(betaIJ) + alphaByLambda * l2.logP(betaIJ);
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeDouble(alphaByLambda);
+ l1.write(out);
+ l2.write(out);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ alphaByLambda = in.readDouble();
+ l1 = new L1();
+ l1.readFields(in);
+ l2 = new L2();
+ l2.readFields(in);
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L1.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L1.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L1.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L1.java Thu Dec 30 02:30:03 2010
@@ -17,6 +17,10 @@
package org.apache.mahout.classifier.sgd;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
/**
* Implements the Laplacian or bi-exponential prior. This prior has a strong tendency to set coefficients to zero
* and thus is useful as an alternative to variable selection. This version implements truncation which prevents
@@ -42,4 +46,14 @@ public class L1 implements PriorFunction
public double logP(double betaIJ) {
return -Math.abs(betaIJ);
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ // stateless class has nothing to serialize
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ // stateless class has nothing to serialize
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L2.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L2.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L2.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/L2.java Thu Dec 30 02:30:03 2010
@@ -17,19 +17,26 @@
package org.apache.mahout.classifier.sgd;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
/**
* Implements the Gaussian prior. This prior has a tendency to decrease large coefficients toward zero, but
* doesn't tend to set them to exactly zero.
*/
public class L2 implements PriorFunction {
- private final double s2;
- private final double s;
+ private double s2;
+ private double s;
public L2(double scale) {
this.s = scale;
this.s2 = scale * scale;
}
+ public L2() {
+ }
+
@Override
public double age(double oldValue, double generations, double learningRate) {
return oldValue * Math.pow(1.0 - learningRate / s2, generations);
@@ -39,4 +46,16 @@ public class L2 implements PriorFunction
public double logP(double betaIJ) {
return -betaIJ * betaIJ / s2 / 2.0 - Math.log(s) - Math.log(2.0 * Math.PI) / 2.0;
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeDouble(s2);
+ out.writeDouble(s);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ s2 = in.readDouble();
+ s = in.readDouble();
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ModelSerializer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ModelSerializer.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ModelSerializer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/ModelSerializer.java Thu Dec 30 02:30:03 2010
@@ -29,7 +29,7 @@ import com.google.gson.JsonPrimitive;
import com.google.gson.JsonSerializationContext;
import com.google.gson.JsonSerializer;
import com.google.gson.reflect.TypeToken;
-import org.apache.mahout.classifier.AbstractVectorClassifier;
+import org.apache.hadoop.io.Writable;
import org.apache.mahout.classifier.OnlineLearner;
import org.apache.mahout.ep.EvolutionaryProcess;
import org.apache.mahout.ep.Mapping;
@@ -40,8 +40,14 @@ import org.apache.mahout.math.Matrix;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.stats.OnlineAuc;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
+import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.lang.reflect.Type;
@@ -98,10 +104,26 @@ public final class ModelSerializer {
* @param clazz The class of the object we expect to read.
* @return The LogisticModelParameters object that we read.
*/
- public static AbstractVectorClassifier loadJsonFrom(Reader in, Class<? extends AbstractVectorClassifier> clazz) {
+ public static <T> T loadJsonFrom(Reader in, Class<T> clazz) {
return gson().fromJson(in, clazz);
}
+ public static void writeBinary(String path, CrossFoldLearner model) throws IOException {
+ PolymorphicWritable.write(new DataOutputStream(new FileOutputStream(path)), model);
+ }
+
+ public static void writeBinary(String path, OnlineLogisticRegression model) throws IOException {
+ PolymorphicWritable.write(new DataOutputStream(new FileOutputStream(path)), model);
+ }
+
+ public static void writeBinary(String path, AdaptiveLogisticRegression model) throws IOException {
+ PolymorphicWritable.write(new DataOutputStream(new FileOutputStream(path)), model);
+ }
+
+ public static <T extends Writable> T readBinary(InputStream in, Class<T> clazz) throws IOException {
+ return PolymorphicWritable.read(new DataInputStream(in), clazz);
+ }
+
private static class PolymorphicTypeAdapter<T> implements JsonDeserializer<T>, JsonSerializer<T> {
@Override
public T deserialize(JsonElement jsonElement,
@@ -162,7 +184,7 @@ public final class ModelSerializer {
new AdaptiveLogisticRegression(x.get("numCategories").getAsInt(),
x.get("numFeatures").getAsInt(),
jdc.<PriorFunction>deserialize(x.get("prior"), PriorFunction.class));
- Type stateType = new TypeToken<State<AdaptiveLogisticRegression.Wrapper>>() {}.getType();
+ Type stateType = new TypeToken<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>() {}.getType();
if (x.get("evaluationInterval")!=null) {
r.setInterval(x.get("evaluationInterval").getAsInt());
} else {
@@ -170,11 +192,11 @@ public final class ModelSerializer {
}
r.setRecord(x.get("record").getAsInt());
- Type epType = new TypeToken<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>>() {}.getType();
- r.setEp(jdc.<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>>deserialize(x.get("ep"), epType));
- r.setSeed(jdc.<State<AdaptiveLogisticRegression.Wrapper>>deserialize(x.get("seed"), stateType));
+ Type epType = new TypeToken<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>() {}.getType();
+ r.setEp(jdc.<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>deserialize(x.get("ep"), epType));
+ r.setSeed(jdc.<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>deserialize(x.get("seed"), stateType));
if (x.get("best") != null) {
- r.setBest(jdc.<State<AdaptiveLogisticRegression.Wrapper>>deserialize(x.get("best"), stateType));
+ r.setBest(jdc.<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>deserialize(x.get("best"), stateType));
}
if (x.get("buffer") != null) {
@@ -189,10 +211,10 @@ public final class ModelSerializer {
public JsonElement serialize(AdaptiveLogisticRegression x, Type type, JsonSerializationContext jsc) {
JsonObject r = new JsonObject();
r.add("ep", jsc.serialize(x.getEp(),
- new TypeToken<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>>() {}.getType()));
+ new TypeToken<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>() {}.getType()));
r.add("minInterval", jsc.serialize(x.getMinInterval()));
r.add("maxInterval", jsc.serialize(x.getMaxInterval()));
- Type stateType = new TypeToken<State<AdaptiveLogisticRegression.Wrapper>>() {}.getType();
+ Type stateType = new TypeToken<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>() {}.getType();
r.add("best", jsc.serialize(x.getBest(), stateType));
r.add("numFeatures", jsc.serialize(x.getNumFeatures()));
r.add("numCategories", jsc.serialize(x.getNumCategories()));
@@ -286,16 +308,16 @@ public final class ModelSerializer {
}
}
- private static class StateTypeAdapter implements JsonSerializer<State<AdaptiveLogisticRegression.Wrapper>>,
- JsonDeserializer<State<AdaptiveLogisticRegression.Wrapper>> {
+ private static class StateTypeAdapter implements JsonSerializer<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>,
+ JsonDeserializer<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>> {
@Override
- public State<AdaptiveLogisticRegression.Wrapper> deserialize(
+ public State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> deserialize(
JsonElement jsonElement, Type type, JsonDeserializationContext jsonDeserializationContext) {
JsonObject v = (JsonObject) jsonElement;
double[] params = asArray(v, "params");
double omni = v.get("omni").getAsDouble();
- State<AdaptiveLogisticRegression.Wrapper> r = new State<AdaptiveLogisticRegression.Wrapper>(params, omni);
+ State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> r = new State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>(params, omni);
double[] step = asArray(v, "step");
r.setId(v.get("id").getAsInt());
@@ -313,7 +335,7 @@ public final class ModelSerializer {
}
@Override
- public JsonElement serialize(State<AdaptiveLogisticRegression.Wrapper> state,
+ public JsonElement serialize(State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> state,
Type type,
JsonSerializationContext jsonSerializationContext) {
JsonObject r = new JsonObject();
@@ -339,41 +361,41 @@ public final class ModelSerializer {
}
private static class EvolutionaryProcessTypeAdapter implements
- InstanceCreator<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>>,
- JsonDeserializer<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>>,
- JsonSerializer<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>> {
- private static final Type STATE_TYPE = new TypeToken<State<AdaptiveLogisticRegression.Wrapper>>() {}.getType();
+ InstanceCreator<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>,
+ JsonDeserializer<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>,
+ JsonSerializer<EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>> {
+ private static final Type STATE_TYPE = new TypeToken<State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>>() {}.getType();
@Override
- public EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper> createInstance(Type type) {
- return new EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>();
+ public EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> createInstance(Type type) {
+ return new EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>();
}
@Override
- public EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper> deserialize(
+ public EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> deserialize(
JsonElement jsonElement, Type type, JsonDeserializationContext jsonDeserializationContext) {
JsonObject x = (JsonObject) jsonElement;
int threadCount = x.get("threadCount").getAsInt();
- EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper> r =
- new EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper>();
+ EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> r =
+ new EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner>();
r.setThreadCount(threadCount);
for (JsonElement element : x.get("population").getAsJsonArray()) {
- State<AdaptiveLogisticRegression.Wrapper> state = jsonDeserializationContext.deserialize(element, STATE_TYPE);
+ State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> state = jsonDeserializationContext.deserialize(element, STATE_TYPE);
r.add(state);
}
return r;
}
@Override
- public JsonElement serialize(EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper> x,
+ public JsonElement serialize(EvolutionaryProcess<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> x,
Type type,
JsonSerializationContext jsc) {
JsonObject r = new JsonObject();
r.add("threadCount", new JsonPrimitive(x.getThreadCount()));
JsonArray v = new JsonArray();
- for (State<AdaptiveLogisticRegression.Wrapper> state : x.getPopulation()) {
+ for (State<AdaptiveLogisticRegression.Wrapper, CrossFoldLearner> state : x.getPopulation()) {
v.add(jsc.serialize(state, STATE_TYPE));
}
r.add("population", v);
@@ -390,4 +412,8 @@ public final class ModelSerializer {
}
return params;
}
+
+ public static void main(String[] args) throws FileNotFoundException {
+ OnlineLogisticRegression m = ModelSerializer.loadJsonFrom(new FileReader("/tmp/news-group-1000.model"), OnlineLogisticRegression.class);
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/OnlineLogisticRegression.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/OnlineLogisticRegression.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/OnlineLogisticRegression.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/OnlineLogisticRegression.java Thu Dec 30 02:30:03 2010
@@ -17,14 +17,23 @@
package org.apache.mahout.classifier.sgd;
+import org.apache.hadoop.io.Writable;
import org.apache.mahout.math.DenseMatrix;
import org.apache.mahout.math.DenseVector;
+import org.apache.mahout.math.MatrixWritable;
+import org.apache.mahout.math.VectorWritable;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
/**
* Extends the basic on-line logistic regression learner with a specific set of learning
* rate annealing schedules.
*/
-public class OnlineLogisticRegression extends AbstractOnlineLogisticRegression {
+public class OnlineLogisticRegression extends AbstractOnlineLogisticRegression implements Writable {
+ public static final int WRITABLE_VERSION = 1;
+
// these next two control decayFactor^steps exponential type of annealing
// learning rate and decay factor
private double mu0 = 1;
@@ -38,7 +47,7 @@ public class OnlineLogisticRegression ex
// controls how per term annealing works
private int perTermAnnealingOffset = 20;
- private OnlineLogisticRegression() {
+ public OnlineLogisticRegression() {
// private constructor available for Gson, but not normal use
}
@@ -121,4 +130,41 @@ public class OnlineLogisticRegression ex
r.copyFrom(this);
return r;
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(WRITABLE_VERSION);
+ out.writeDouble(mu0);
+ out.writeDouble(decayFactor);
+ out.writeInt(stepOffset);
+ out.writeInt(step);
+ out.writeDouble(forgettingExponent);
+ out.writeInt(perTermAnnealingOffset);
+ out.writeInt(numCategories);
+ MatrixWritable.writeMatrix(out, beta);
+ PolymorphicWritable.write(out, prior);
+ VectorWritable.writeVector(out, updateCounts);
+ VectorWritable.writeVector(out, updateSteps);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ int version = in.readInt();
+ if (version == WRITABLE_VERSION) {
+ mu0 = in.readDouble();
+ decayFactor = in.readDouble();
+ stepOffset = in.readInt();
+ step = in.readInt();
+ forgettingExponent = in.readDouble();
+ perTermAnnealingOffset = in.readInt();
+ numCategories = in.readInt();
+ beta = MatrixWritable.readMatrix(in);
+ prior = PolymorphicWritable.read(in, PriorFunction.class);
+
+ updateCounts = VectorWritable.readVector(in);
+ updateSteps = VectorWritable.readVector(in);
+ } else {
+ throw new IOException("Incorrect object version, wanted " + WRITABLE_VERSION + " got " + version);
+ }
+ }
}
Added: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PolymorphicWritable.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PolymorphicWritable.java?rev=1053787&view=auto
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PolymorphicWritable.java (added)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PolymorphicWritable.java Thu Dec 30 02:30:03 2010
@@ -0,0 +1,50 @@
+/*
+ * 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.classifier.sgd;
+
+import org.apache.hadoop.io.Writable;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Utilities that write a class name and then serialize using writables.
+ */
+public class PolymorphicWritable<T> {
+ public static <T extends Writable> void write(DataOutput dataOutput, T value) throws IOException {
+ dataOutput.writeUTF(value.getClass().getName());
+ value.write(dataOutput);
+ }
+
+ public static <T extends Writable> T read(DataInput dataInput, Class<? extends T> clazz) throws IOException {
+ String className = dataInput.readUTF();
+ T r = null;
+ try {
+ r = Class.forName(className).asSubclass(clazz).newInstance();
+ } catch (InstantiationException e) {
+ throw new IOException("Can't create object", e);
+ } catch (IllegalAccessException e) {
+ throw new IOException("Can't access constructor", e);
+ } catch (ClassNotFoundException e) {
+ throw new IOException("No such class", e);
+ }
+ r.readFields(dataInput);
+ return r;
+ }
+}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PriorFunction.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PriorFunction.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PriorFunction.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/PriorFunction.java Thu Dec 30 02:30:03 2010
@@ -17,6 +17,8 @@
package org.apache.mahout.classifier.sgd;
+import org.apache.hadoop.io.Writable;
+
/**
* A prior is used to regularize the learning algorithm. This allows a trade-off to
* be made between complexity of the model being learned and the accuracy with which
@@ -24,7 +26,7 @@ package org.apache.mahout.classifier.sgd
* which can be approximated using different priors. For large sparse systems, such
* as text classification, the L1 prior is often used which favors sparse models.
*/
-public interface PriorFunction {
+public interface PriorFunction extends Writable {
/**
* Applies the regularization to a coefficient.
* @param oldValue The previous value.
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/TPrior.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/TPrior.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/TPrior.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/TPrior.java Thu Dec 30 02:30:03 2010
@@ -19,11 +19,15 @@ package org.apache.mahout.classifier.sgd
import org.apache.commons.math.special.Gamma;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
/**
* Provides a t-distribution as a prior.
*/
public class TPrior implements PriorFunction {
- private final double df;
+ private double df;
public TPrior(double df) {
this.df = df;
@@ -44,4 +48,14 @@ public class TPrior implements PriorFunc
- Gamma.logGamma(df / 2.0)
- (df + 1.0) / 2.0 * Math.log(1.0 + betaIJ * betaIJ);
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeDouble(df);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ df = in.readDouble();
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/UniformPrior.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/UniformPrior.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/UniformPrior.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/sgd/UniformPrior.java Thu Dec 30 02:30:03 2010
@@ -17,6 +17,10 @@
package org.apache.mahout.classifier.sgd;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
/**
* A uniform prior. This is an improper prior that corresponds to no regularization at all.
*/
@@ -30,4 +34,14 @@ public class UniformPrior implements Pri
public double logP(double betaIJ) {
return 0;
}
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ // nothing to write
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ // stateless class is trivial to read
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/ep/EvolutionaryProcess.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/ep/EvolutionaryProcess.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/ep/EvolutionaryProcess.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/ep/EvolutionaryProcess.java Thu Dec 30 02:30:03 2010
@@ -18,7 +18,12 @@
package org.apache.mahout.ep;
import com.google.common.collect.Lists;
+import org.apache.hadoop.io.Writable;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -59,7 +64,7 @@ import java.util.concurrent.Future;
*
* @param <T> The payload class.
*/
-public class EvolutionaryProcess<T extends Payload<T>> {
+public class EvolutionaryProcess<T extends Payload<U>, U> implements Writable {
// used to execute operations on the population in thread parallel.
private transient ExecutorService pool;
@@ -67,7 +72,7 @@ public class EvolutionaryProcess<T exten
private int threadCount;
// list of members of the population
- private List<State<T>> population;
+ private List<State<T, U>> population;
// how big should the population be. If this is changed, it will take effect
// the next time the population is mutated.
@@ -85,20 +90,20 @@ public class EvolutionaryProcess<T exten
* @param populationSize How large a population to use
* @param seed An initial population member
*/
- public EvolutionaryProcess(int threadCount, int populationSize, State<T> seed) {
+ public EvolutionaryProcess(int threadCount, int populationSize, State<T, U> seed) {
this.populationSize = populationSize;
setThreadCount(threadCount);
initializePopulation(populationSize, seed);
}
- private void initializePopulation(int populationSize, State<T> seed) {
+ private void initializePopulation(int populationSize, State<T, U> seed) {
population = Lists.newArrayList(seed);
for (int i = 0; i < populationSize; i++) {
population.add(seed.mutate());
}
}
- public void add(State<T> value) {
+ public void add(State<T, U> value) {
population.add(value);
}
@@ -112,7 +117,7 @@ public class EvolutionaryProcess<T exten
Collections.sort(population);
// we copy here to avoid concurrent modification
- List<State<T>> parents = Lists.newArrayList(population.subList(0, survivors));
+ List<State<T, U>> parents = Lists.newArrayList(population.subList(0, survivors));
population.subList(survivors, population.size()).clear();
// fill out the population with offspring from the survivors
@@ -132,12 +137,12 @@ public class EvolutionaryProcess<T exten
* @throws ExecutionException If fn throws an exception, that exception will be collected
* and rethrown nested in an ExecutionException.
*/
- public State<T> parallelDo(final Function<T> fn) throws InterruptedException, ExecutionException {
- Collection<Callable<State<T>>> tasks = Lists.newArrayList();
- for (final State<T> state : population) {
- tasks.add(new Callable<State<T>>() {
+ public State<T, U> parallelDo(final Function<Payload<U>> fn) throws InterruptedException, ExecutionException {
+ Collection<Callable<State<T, U>>> tasks = Lists.newArrayList();
+ for (final State<T, U> state : population) {
+ tasks.add(new Callable<State<T, U>>() {
@Override
- public State<T> call() {
+ public State<T, U> call() {
double v = fn.apply(state.getPayload(), state.getMappedParams());
state.setValue(v);
return state;
@@ -145,13 +150,13 @@ public class EvolutionaryProcess<T exten
});
}
- List<Future<State<T>>> r = pool.invokeAll(tasks);
+ List<Future<State<T, U>>> r = pool.invokeAll(tasks);
// zip through the results and find the best one
double max = Double.NEGATIVE_INFINITY;
- State<T> best = null;
- for (Future<State<T>> future : r) {
- State<T> s = future.get();
+ State<T, U> best = null;
+ for (Future<State<T, U>> future : r) {
+ State<T, U> s = future.get();
double value = s.getValue();
if (!Double.isNaN(value) && value >= max) {
max = value;
@@ -178,7 +183,7 @@ public class EvolutionaryProcess<T exten
return populationSize;
}
- public List<State<T>> getPopulation() {
+ public List<State<T, U>> getPopulation() {
return population;
}
@@ -198,4 +203,24 @@ public class EvolutionaryProcess<T exten
super(message);
}
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(threadCount);
+ out.writeInt(population.size());
+ for (State<T, U> state : population) {
+ PolymorphicWritable.write(out, state);
+ }
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ setThreadCount(input.readInt());
+ int n = input.readInt();
+ population = Lists.newArrayList();
+ for (int i = 0; i < n; i++) {
+ State<T, U> state = PolymorphicWritable.read(input, State.class);
+ population.add(state);
+ }
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/ep/Mapping.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/ep/Mapping.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/ep/Mapping.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/ep/Mapping.java Thu Dec 30 02:30:03 2010
@@ -17,15 +17,20 @@
package org.apache.mahout.ep;
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.io.Writable;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.mahout.math.function.UnaryFunction;
-import com.google.common.base.Preconditions;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
/**
* Provides coordinate tranformations so that evolution can proceed on the entire space of
* reals but have the output limited and squished in convenient (and safe) ways.
*/
-public abstract class Mapping implements UnaryFunction {
+public abstract class Mapping implements UnaryFunction, Writable {
private Mapping() {
}
@@ -35,7 +40,7 @@ public abstract class Mapping implements
private double max;
private double scale;
- private SoftLimit() {
+ public SoftLimit() {
}
private SoftLimit(double min, double max, double scale) {
@@ -49,12 +54,25 @@ public abstract class Mapping implements
return min + (max - min) * 1 / (1 + Math.exp(-v * scale));
}
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeDouble(min);
+ out.writeDouble(max);
+ out.writeDouble(scale);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ min = in.readDouble();
+ max = in.readDouble();
+ scale = in.readDouble();
+ }
}
public static final class LogLimit extends Mapping {
private Mapping wrapped;
- private LogLimit() {
+ public LogLimit() {
}
private LogLimit(double low, double high) {
@@ -65,12 +83,22 @@ public abstract class Mapping implements
public double apply(double v) {
return Math.exp(wrapped.apply(v));
}
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ PolymorphicWritable.write(dataOutput, wrapped);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ wrapped = PolymorphicWritable.read(in, Mapping.class);
+ }
}
public static final class Exponential extends Mapping {
private double scale;
- private Exponential() {
+ public Exponential() {
}
private Exponential(double scale) {
@@ -81,16 +109,36 @@ public abstract class Mapping implements
public double apply(double v) {
return Math.exp(v * scale);
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeDouble(scale);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ scale = in.readDouble();
+ }
}
public static final class Identity extends Mapping {
- private Identity() {
+ public Identity() {
}
@Override
public double apply(double v) {
return v;
}
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ // stateless
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ // stateless
+ }
}
/**
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/ep/Payload.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/ep/Payload.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/ep/Payload.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/ep/Payload.java Thu Dec 30 02:30:03 2010
@@ -17,18 +17,20 @@
package org.apache.mahout.ep;
+import org.apache.hadoop.io.Writable;
+
/**
- * Payloads for evolutionary state must be copyable and updatable. The copy should be
- * a deep copy unless some aspect of the state is sharable or immutable.
- *
- * During mutation, a copy is first made and then after the parameters in the State
- * structure are suitably modified, update is called with the scaled versions of the
- * parameters.
+ * Payloads for evolutionary state must be copyable and updatable. The copy should be a deep copy
+ * unless some aspect of the state is sharable or immutable.
+ * <p/>
+ * During mutation, a copy is first made and then after the parameters in the State structure are
+ * suitably modified, update is called with the scaled versions of the parameters.
*
- * @see State
* @param <T>
+ * @see State
*/
-public interface Payload<T> {
- T copy();
+public interface Payload<T> extends Writable {
+ Payload<T> copy();
+
void update(double[] params);
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/ep/State.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/ep/State.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/ep/State.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/ep/State.java Thu Dec 30 02:30:03 2010
@@ -18,8 +18,13 @@
package org.apache.mahout.ep;
import com.google.common.collect.Lists;
+import org.apache.hadoop.io.Writable;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
import org.apache.mahout.common.RandomUtils;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Locale;
@@ -44,7 +49,7 @@ import java.util.concurrent.atomic.Atomi
*
* @see Mapping
*/
-public class State<T extends Payload<T>> implements Comparable<State<T>> {
+public class State<T extends Payload<U>, U> implements Comparable<State<T, U>>, Writable {
// object count is kept to break ties in comparison.
private static final AtomicInteger OBJECT_COUNT = new AtomicInteger();
@@ -61,7 +66,7 @@ public class State<T extends Payload<T>>
private double[] step;
// current fitness value
private double value;
- private T payload;
+ private Payload<U> payload;
public State() {
}
@@ -79,8 +84,8 @@ public class State<T extends Payload<T>>
/**
* Deep copies a state, useful in mutation.
*/
- public State<T> copy() {
- State<T> r = new State<T>();
+ public State<T, U> copy() {
+ State<T, U> r = new State<T, U>();
r.params = Arrays.copyOf(this.params, this.params.length);
r.omni = this.omni;
r.step = Arrays.copyOf(this.step, this.step.length);
@@ -98,7 +103,7 @@ public class State<T extends Payload<T>>
*
* @return A new state.
*/
- public State<T> mutate() {
+ public State<T, U> mutate() {
double sum = 0;
for (double v : step) {
sum += v * v;
@@ -106,7 +111,7 @@ public class State<T extends Payload<T>>
sum = Math.sqrt(sum);
double lambda = 1 + gen.nextGaussian();
- State<T> r = this.copy();
+ State<T, U> r = this.copy();
double magnitude = 0.9 * omni + sum / 10;
r.omni = magnitude * -Math.log(1 - gen.nextDouble());
for (int i = 0; i < step.length; i++) {
@@ -172,7 +177,7 @@ public class State<T extends Payload<T>>
}
public T getPayload() {
- return payload;
+ return (T) payload;
}
public double getValue() {
@@ -213,7 +218,7 @@ public class State<T extends Payload<T>>
if (!(o instanceof State)) {
return false;
}
- State<?> other = (State<?>) o;
+ State<?,?> other = (State<?,?>) o;
return id == other.id && value == other.value;
}
@@ -229,7 +234,7 @@ public class State<T extends Payload<T>>
* @return -1, 0, 1 if the other state is better, identical or worse than this one.
*/
@Override
- public int compareTo(State<T> other) {
+ public int compareTo(State<T, U> other) {
int r = Double.compare(other.value, this.value);
if (r != 0) {
return r;
@@ -249,4 +254,46 @@ public class State<T extends Payload<T>>
}
return String.format(Locale.ENGLISH, "<S/%s %.3f %.3f>", payload, omni + Math.sqrt(sum), value);
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(id);
+ out.writeInt(params.length);
+ for (double v : params) {
+ out.writeDouble(v);
+ }
+ for (Mapping map : maps) {
+ PolymorphicWritable.write(out, map);
+ }
+
+ out.writeDouble(omni);
+ for (double v : step) {
+ out.writeDouble(v);
+ }
+
+ out.writeDouble(value);
+ PolymorphicWritable.write(out, payload);
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ id = input.readInt();
+ int n = input.readInt();
+ params = new double[n];
+ for (int i = 0; i < n; i++) {
+ params[i] = input.readDouble();
+ }
+
+ maps = new Mapping[n];
+ for (int i = 0; i < n; i++) {
+ maps[i] = PolymorphicWritable.read(input, Mapping.class);
+ }
+ omni = input.readDouble();
+ step = new double[n];
+ for (int i = 0; i < n; i++) {
+ step[i] = input.readDouble();
+ }
+ value = input.readDouble();
+ payload = PolymorphicWritable.read(input, Payload.class);
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/ep/ThreadedEvolutionaryProcess.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/ep/ThreadedEvolutionaryProcess.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/ep/ThreadedEvolutionaryProcess.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/ep/ThreadedEvolutionaryProcess.java Thu Dec 30 02:30:03 2010
@@ -17,12 +17,12 @@
package org.apache.mahout.ep;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import java.util.Deque;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.Locale;
import java.util.Map;
import java.util.PriorityQueue;
@@ -41,37 +41,37 @@ import java.util.concurrent.Future;
*/
public class ThreadedEvolutionaryProcess {
- private static final PriorityQueue<State<?>> resultPopulation = new PriorityQueue<State<?>>();
+ private static final PriorityQueue<State<?, ?>> resultPopulation = new PriorityQueue<State<?, ?>>();
private volatile int taskCount;
private volatile int processCount;
private volatile int maxTask;
- private final Deque<State<?>> pending = new LinkedList<State<?>>();
- private final Set<Future<State<?>>> working = Sets.newHashSet();
+ private final Deque<State<?, ?>> pending = Lists.newLinkedList();
+ private final Set<Future<State<?,?>>> working = Sets.newHashSet();
private final ExecutorService pool;
- private final ExecutorCompletionService<State<?>> ecs;
+ private final ExecutorCompletionService<State<?, ?>> ecs;
private final int threadCount;
private final Map<Integer, Mapping> mappingTable = Maps.newHashMap();
public ThreadedEvolutionaryProcess(int threadCount) {
this.threadCount = threadCount;
pool = Executors.newFixedThreadPool(threadCount);
- ecs = new ExecutorCompletionService<State<?>>(pool);
+ ecs = new ExecutorCompletionService<State<?, ?>>(pool);
}
public void setMap(int i, Mapping m) {
mappingTable.put(i, m);
}
- public State<?> optimize(Function f, int dim, long timeLimit, int parentDepth)
+ public State<?, ?> optimize(Function f, int dim, long timeLimit, int parentDepth)
throws InterruptedException, ExecutionException {
long t0 = System.currentTimeMillis();
// start with a few points near 0. These will get transformed
- State<?> s0 = new State(new double[dim], 0.5);
+ State<?, ?> s0 = new State(new double[dim], 0.5);
for (Map.Entry<Integer, Mapping> entry : mappingTable.entrySet()) {
s0.setMap(entry.getKey(), entry.getValue());
}
@@ -85,37 +85,37 @@ public class ThreadedEvolutionaryProcess
do {
// launch new tasks until we fill the available slots
while (working.size() < threadCount && !pending.isEmpty()) {
- State<?> next = pending.removeFirst();
+ State<?, ?> next = pending.removeFirst();
working.add(ecs.submit(new EvalTask(f, next)));
processCount++;
}
// wait for at least one result, then grab any additional results
- Future<State<?>> result = ecs.take();
+ Future<State<?, ?>> result = ecs.take();
while (result != null) {
- State<?> r = result.get();
+ State<?, ?> r = result.get();
resultPopulation.add(r);
working.remove(result);
result = ecs.poll();
}
// now spawn new pending tasks from the best in recent history
- State<?>[] parents = new State[parentDepth];
- Iterator<State<?>> j = resultPopulation.iterator();
+ State<?, ?>[] parents = new State[parentDepth];
+ Iterator<State<?, ?>> j = resultPopulation.iterator();
for (int i = 0; i < parentDepth && j.hasNext(); i++) {
parents[i] = j.next();
}
int k = 0;
while (pending.size() + working.size() < threadCount) {
- State<?> tmp = parents[(k++) % parentDepth];
+ State<?, ?> tmp = parents[(k++) % parentDepth];
pending.add(tmp.mutate());
}
} while (System.currentTimeMillis() - t0 < timeLimit);
// wait for last results to dribble in
while (!working.isEmpty()) {
- Future<State<?>> result = ecs.take();
+ Future<State<?, ?>> result = ecs.take();
working.remove(result);
resultPopulation.add(result.get());
}
@@ -131,11 +131,11 @@ public class ThreadedEvolutionaryProcess
return String.format(Locale.ENGLISH, "Launched %d function evaluations\nMaximum threading width was %d", processCount, maxTask);
}
- public class EvalTask implements Callable<State<?>> {
+ public class EvalTask implements Callable<State<?, ?>> {
private final Function f;
- private final State<?> what;
+ private final State<?, ?> what;
- public EvalTask(Function f, State<?> what) {
+ public EvalTask(Function f, State<?, ?> what) {
this.f = f;
this.what = what;
}
@@ -146,7 +146,7 @@ public class ThreadedEvolutionaryProcess
* @return computed result
*/
@Override
- public State<?> call() {
+ public State<?, ?> call() {
taskCount++;
maxTask = Math.max(taskCount, maxTask);
try {
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/MatrixWritable.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/MatrixWritable.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/MatrixWritable.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/MatrixWritable.java Thu Dec 30 02:30:03 2010
@@ -17,6 +17,7 @@
package org.apache.mahout.math;
+import com.google.common.base.Preconditions;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
@@ -27,6 +28,9 @@ import java.util.Map;
public class MatrixWritable implements Writable {
private Matrix matrix;
+ private static final int NUM_FLAGS = 2;
+ private static final int FLAG_DENSE = 1;
+ private static final int FLAG_SEQUENTIAL = 2;
public MatrixWritable() {
}
@@ -95,25 +99,45 @@ public class MatrixWritable implements W
/** Reads a typed Matrix instance from the input stream */
public static Matrix readMatrix(DataInput in) throws IOException {
- String matrixClassName = in.readUTF();
- Matrix matrix;
- try {
- matrix = Class.forName(matrixClassName).asSubclass(Matrix.class)
- .newInstance();
- } catch (ClassNotFoundException e) {
- throw new IllegalStateException(e);
- } catch (IllegalAccessException e) {
- throw new IllegalStateException(e);
- } catch (InstantiationException e) {
- throw new IllegalStateException(e);
+ int flags = in.readInt();
+ Preconditions.checkArgument(flags >> NUM_FLAGS == 0, "Unknown flags set: %d", Integer.toString(flags, 2));
+ boolean dense = (flags & FLAG_DENSE) != 0;
+ boolean sequential = (flags & FLAG_SEQUENTIAL) != 0;
+
+ int rows = in.readInt();
+ int columns = in.readInt();
+
+ Matrix r ;
+ if (dense) {
+ r = new DenseMatrix(rows, columns);
+ } else {
+ r = new SparseRowMatrix(new int[]{rows, columns}, !sequential);
}
- // matrix.readFields(in);
- return matrix;
+
+ for (int row = 0; row < rows; row++) {
+ r.viewRow(row).assign(VectorWritable.readVector(in));
+ }
+
+ return r;
}
/** Writes a typed Matrix instance to the output stream */
public static void writeMatrix(DataOutput out, Matrix matrix) throws IOException {
- out.writeUTF(matrix.getClass().getName());
- // matrix.write(out);
+ int flags = 0;
+ Vector row = matrix.viewRow(0);
+ if (row.isDense()) {
+ flags |= FLAG_DENSE;
+ }
+ if (row.isSequentialAccess()) {
+ flags |= FLAG_SEQUENTIAL;
+ }
+ out.writeInt(flags);
+
+ out.writeInt(matrix.rowSize());
+ out.writeInt(matrix.columnSize());
+
+ for (int i = 0; i < matrix.rowSize(); i++) {
+ VectorWritable.writeVector(out, matrix.viewRow(i), false);
+ }
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/VectorWritable.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/VectorWritable.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/VectorWritable.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/VectorWritable.java Thu Dec 30 02:30:03 2010
@@ -72,58 +72,7 @@ public final class VectorWritable extend
@Override
public void write(DataOutput out) throws IOException {
-
- boolean dense = vector.isDense();
- boolean sequential = vector.isSequentialAccess();
- boolean named = vector instanceof NamedVector;
-
- boolean writesLaxPrecision = this.writesLaxPrecision;
- out.writeByte((dense ? FLAG_DENSE : 0)
- | (sequential ? FLAG_SEQUENTIAL : 0)
- | (named ? FLAG_NAMED : 0)
- | (writesLaxPrecision ? FLAG_LAX_PRECISION : 0));
-
- Varint.writeUnsignedVarInt(vector.size(), out);
- if (dense) {
- for (Vector.Element element : vector) {
- if (writesLaxPrecision) {
- out.writeFloat((float) element.get());
- } else {
- out.writeDouble(element.get());
- }
- }
- } else {
- Varint.writeUnsignedVarInt(vector.getNumNondefaultElements(), out);
- Iterator<Vector.Element> iter = vector.iterateNonZero();
- if (sequential) {
- int lastIndex = 0;
- while (iter.hasNext()) {
- Vector.Element element = iter.next();
- int thisIndex = element.index();
- // Delta-code indices:
- Varint.writeUnsignedVarInt(thisIndex - lastIndex, out);
- lastIndex = thisIndex;
- if (writesLaxPrecision) {
- out.writeFloat((float) element.get());
- } else {
- out.writeDouble(element.get());
- }
- }
- } else {
- while (iter.hasNext()) {
- Vector.Element element = iter.next();
- Varint.writeUnsignedVarInt(element.index(), out);
- if (writesLaxPrecision) {
- out.writeFloat((float) element.get());
- } else {
- out.writeDouble(element.get());
- }
- }
- }
- }
- if (named) {
- out.writeUTF(((NamedVector) vector).getName());
- }
+ writeVector(out, this.vector, this.writesLaxPrecision);
}
@Override
@@ -174,7 +123,65 @@ public final class VectorWritable extend
/** Write the vector to the output */
public static void writeVector(DataOutput out, Vector vector) throws IOException {
- new VectorWritable(vector).write(out);
+ writeVector(out, vector, false);
}
+
+ public static void writeVector(DataOutput out, Vector vector, boolean laxPrecision) throws IOException {
+ boolean dense = vector.isDense();
+ boolean sequential = vector.isSequentialAccess();
+ boolean named = vector instanceof NamedVector;
+
+ out.writeByte((dense ? FLAG_DENSE : 0)
+ | (sequential ? FLAG_SEQUENTIAL : 0)
+ | (named ? FLAG_NAMED : 0)
+ | (laxPrecision ? FLAG_LAX_PRECISION : 0));
+ Varint.writeUnsignedVarInt(vector.size(), out);
+ if (dense) {
+ for (Vector.Element element : vector) {
+ if (laxPrecision) {
+ out.writeFloat((float) element.get());
+ } else {
+ out.writeDouble(element.get());
+ }
+ }
+ } else {
+ Varint.writeUnsignedVarInt(vector.getNumNondefaultElements(), out);
+ Iterator<Vector.Element> iter = vector.iterateNonZero();
+ if (sequential) {
+ int lastIndex = 0;
+ while (iter.hasNext()) {
+ Vector.Element element = iter.next();
+ int thisIndex = element.index();
+ // Delta-code indices:
+ Varint.writeUnsignedVarInt(thisIndex - lastIndex, out);
+ lastIndex = thisIndex;
+ if (laxPrecision) {
+ out.writeFloat((float) element.get());
+ } else {
+ out.writeDouble(element.get());
+ }
+ }
+ } else {
+ while (iter.hasNext()) {
+ Vector.Element element = iter.next();
+ Varint.writeUnsignedVarInt(element.index(), out);
+ if (laxPrecision) {
+ out.writeFloat((float) element.get());
+ } else {
+ out.writeDouble(element.get());
+ }
+ }
+ }
+ }
+ if (named) {
+ out.writeUTF(((NamedVector) vector).getName());
+ }
+ }
+
+ public static Vector readVector(DataInput in) throws IOException {
+ VectorWritable v = new VectorWritable();
+ v.readFields(in);
+ return v.get();
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GlobalOnlineAuc.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GlobalOnlineAuc.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GlobalOnlineAuc.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GlobalOnlineAuc.java Thu Dec 30 02:30:03 2010
@@ -21,8 +21,13 @@ import org.apache.mahout.common.RandomUt
import org.apache.mahout.math.DenseMatrix;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.Matrix;
+import org.apache.mahout.math.MatrixWritable;
import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.VectorWritable;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.Random;
/**
@@ -32,6 +37,7 @@ import java.util.Random;
* batch fashion. The probabilistic definition (the probability that a random element of one set
* has a higher score than a random element of another set) gives us a way to estimate this
* on-line.
+ *
* @see GroupedOnlineAuc
*/
public class GlobalOnlineAuc implements OnlineAuc {
@@ -50,9 +56,9 @@ public class GlobalOnlineAuc implements
// FIFO has distinctly the best properties as a policy. See OnlineAucTest for details
private ReplacementPolicy policy = ReplacementPolicy.FIFO;
private final transient Random random = RandomUtils.getRandom();
- private final Matrix scores;
- private final Vector averages;
- private final Vector samples;
+ private Matrix scores;
+ private Vector averages;
+ private Vector samples;
public GlobalOnlineAuc() {
int numCategories = 2;
@@ -67,7 +73,7 @@ public class GlobalOnlineAuc implements
public double addSample(int category, String groupKey, double score) {
return addSample(category, score);
}
-
+
@Override
public double addSample(int category, double score) {
int n = (int) samples.get(category);
@@ -108,14 +114,14 @@ public class GlobalOnlineAuc implements
count++;
if (score > v) {
m++;
- // } else if (score < v) {
+ // } else if (score < v) {
// m += 0
} else if (score == v) {
m += 0.5;
}
}
averages.set(category, averages.get(category)
- + (m / count - averages.get(category)) / Math.min(windowSize, samples.get(category)));
+ + (m / count - averages.get(category)) / Math.min(windowSize, samples.get(category)));
}
return auc();
}
@@ -139,4 +145,24 @@ public class GlobalOnlineAuc implements
public void setWindowSize(int windowSize) {
this.windowSize = windowSize;
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(windowSize);
+ out.writeInt(policy.ordinal());
+ MatrixWritable.writeMatrix(out, scores);
+ VectorWritable.writeVector(out, averages);
+ VectorWritable.writeVector(out, samples);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ windowSize = in.readInt();
+ policy = ReplacementPolicy.values()[in.readInt()];
+
+ scores = MatrixWritable.readMatrix(in);
+ averages = VectorWritable.readVector(in);
+ samples = VectorWritable.readVector(in);
+ }
+
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GroupedOnlineAuc.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GroupedOnlineAuc.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GroupedOnlineAuc.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/GroupedOnlineAuc.java Thu Dec 30 02:30:03 2010
@@ -18,7 +18,11 @@
package org.apache.mahout.math.stats;
import com.google.common.collect.Maps;
+import org.apache.mahout.classifier.sgd.PolymorphicWritable;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.util.Map;
/**
@@ -83,4 +87,27 @@ public class GroupedOnlineAuc implements
auc.setWindowSize(windowSize);
}
}
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(map.size());
+ for (String key : map.keySet()) {
+ out.writeUTF(key);
+ PolymorphicWritable.write(out, map.get(key));
+ }
+ out.writeInt(policy.ordinal());
+ out.writeInt(windowSize);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ int n = in.readInt();
+ map.clear();
+ for (int i = 0; i < n; i++) {
+ String key = in.readUTF();
+ map.put(key, PolymorphicWritable.read(in, OnlineAuc.class));
+ }
+ policy = GlobalOnlineAuc.ReplacementPolicy.values()[in.readInt()];
+ windowSize = in.readInt();
+ }
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/OnlineAuc.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/OnlineAuc.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/OnlineAuc.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/math/stats/OnlineAuc.java Thu Dec 30 02:30:03 2010
@@ -17,13 +17,15 @@
package org.apache.mahout.math.stats;
+import org.apache.hadoop.io.Writable;
+
/**
* Describes the generic outline of how to compute AUC. Currently there are two
* implementations of this, one for computing a global estimate of AUC and the other
* for computing average grouped AUC. Grouped AUC is useful when misusing a classifier
* as a recommendation system.
*/
-public interface OnlineAuc {
+public interface OnlineAuc extends Writable {
double addSample(int category, String groupKey, double score);
double addSample(int category, double score);
Modified: mahout/trunk/core/src/test/java/org/apache/mahout/classifier/sgd/ModelSerializerTest.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/test/java/org/apache/mahout/classifier/sgd/ModelSerializerTest.java?rev=1053787&r1=1053786&r2=1053787&view=diff
==============================================================================
--- mahout/trunk/core/src/test/java/org/apache/mahout/classifier/sgd/ModelSerializerTest.java (original)
+++ mahout/trunk/core/src/test/java/org/apache/mahout/classifier/sgd/ModelSerializerTest.java Thu Dec 30 02:30:03 2010
@@ -20,6 +20,7 @@ package org.apache.mahout.classifier.sgd
import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
+import org.apache.hadoop.io.Writable;
import org.apache.mahout.classifier.OnlineLearner;
import org.apache.mahout.common.MahoutTestCase;
import org.apache.mahout.common.RandomUtils;
@@ -32,6 +33,11 @@ import org.apache.mahout.math.stats.Glob
import org.apache.mahout.math.stats.OnlineAuc;
import org.junit.Test;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.io.StringReader;
import java.lang.reflect.Type;
import java.util.Iterator;
@@ -41,27 +47,44 @@ import java.util.Random;
public final class ModelSerializerTest extends MahoutTestCase {
@Test
- public void testSoftLimitDeserialization() {
+ public void testSoftLimitDeserialization() throws IOException {
Mapping m = ModelSerializer.gson().fromJson(new StringReader("{\"min\":-18.420680743952367,\"max\":-2.3025850929940455,\"scale\":1.0}"), Mapping.SoftLimit.class);
assertTrue(m instanceof Mapping.SoftLimit);
assertEquals((-18.420680743952367 - 2.3025850929940455) / 2, m.apply(0), 1.0e-6);
+ Mapping m2 = roundTrip(m, Mapping.class);
+ assertTrue(m2 instanceof Mapping.SoftLimit);
+ assertEquals((-18.420680743952367 - 2.3025850929940455) / 2, m2.apply(0), 1.0e-6);
+
String data = "{\"class\":\"org.apache.mahout.ep.Mapping$SoftLimit\",\"value\":{\"min\":-18.420680743952367,\"max\":-2.3025850929940455,\"scale\":1.0}}";
- m = ModelSerializer.gson().fromJson(new StringReader(data), Mapping.class);
+ m = ModelSerializer.loadJsonFrom(new StringReader(data), Mapping.class);
assertTrue(m instanceof Mapping.SoftLimit);
assertEquals((-18.420680743952367 - 2.3025850929940455) / 2, m.apply(0), 1.0e-6);
}
+ private static <T extends Writable> T roundTrip(T m, Class<T> clazz) throws IOException {
+ ByteArrayOutputStream buf = new ByteArrayOutputStream(1000);
+ DataOutputStream dos = new DataOutputStream(buf);
+ PolymorphicWritable.write(dos, m);
+ dos.close();
+
+ return PolymorphicWritable.read(new DataInputStream(new ByteArrayInputStream(buf.toByteArray())), clazz);
+ }
+
@Test
- public void testMappingDeserialization() {
+ public void testMappingDeserialization() throws IOException {
String data = "{\"class\":\"org.apache.mahout.ep.Mapping$LogLimit\",\"value\":{\"wrapped\":{\"class\":\"org.apache.mahout.ep.Mapping$SoftLimit\",\"value\":{\"min\":-18.420680743952367,\"max\":-2.3025850929940455,\"scale\":1.0}}}}";
- Mapping m = ModelSerializer.gson().fromJson(new StringReader(data), Mapping.class);
+ Mapping m = ModelSerializer.loadJsonFrom(new StringReader(data), Mapping.class);
assertTrue(m instanceof Mapping.LogLimit);
assertEquals(Math.sqrt(Math.exp(-18.420680743952367) * Math.exp(-2.3025850929940455)), m.apply(0), 1.0e-6);
+
+ Mapping m2 = roundTrip(m, Mapping.class);
+ assertTrue(m2 instanceof Mapping.LogLimit);
+ assertEquals(Math.sqrt(Math.exp(-18.420680743952367) * Math.exp(-2.3025850929940455)), m2.apply(0), 1.0e-6);
}
@Test
- public void onlineAucRoundtrip() {
+ public void onlineAucRoundtrip() throws IOException {
RandomUtils.useTestSeed();
OnlineAuc auc1 = new GlobalOnlineAuc();
Random gen = RandomUtils.getRandom();
@@ -71,12 +94,13 @@ public final class ModelSerializerTest e
}
assertEquals(0.76, auc1.auc(), 0.01);
- Gson gson = ModelSerializer.gson();
- String s = gson.toJson(auc1);
+ String s = ModelSerializer.gson().toJson(auc1);
- OnlineAuc auc2 = gson.fromJson(s, GlobalOnlineAuc.class);
+ OnlineAuc auc2 = ModelSerializer.gson().fromJson(s, GlobalOnlineAuc.class);
+ OnlineAuc auc3 = roundTrip(auc2, OnlineAuc.class);
assertEquals(auc1.auc(), auc2.auc(), 0);
+ assertEquals(auc1.auc(), auc3.auc(), 0);
for (int i = 0; i < 1000; i++) {
auc1.addSample(0, gen.nextGaussian());
@@ -84,18 +108,22 @@ public final class ModelSerializerTest e
auc2.addSample(0, gen.nextGaussian());
auc2.addSample(1, gen.nextGaussian() + 1);
+
+ auc3.addSample(0, gen.nextGaussian());
+ auc3.addSample(1, gen.nextGaussian() + 1);
}
assertEquals(auc1.auc(), auc2.auc(), 0.01);
+ assertEquals(auc1.auc(), auc3.auc(), 0.01);
Foo x = new Foo();
x.foo = auc1;
x.pig = 3.13;
x.dog = 42;
- s = gson.toJson(x);
+ s = ModelSerializer.gson().toJson(x);
- Foo y = gson.fromJson(s, Foo.class);
+ Foo y = ModelSerializer.gson().fromJson(s, Foo.class);
assertEquals(auc1.auc(), y.foo.auc(), 0.01);
}
@@ -107,55 +135,66 @@ public final class ModelSerializerTest e
}
@Test
- public void onlineLogisticRegressionRoundTrip() {
+ public void onlineLogisticRegressionRoundTrip() throws IOException {
OnlineLogisticRegression olr = new OnlineLogisticRegression(2, 5, new L1());
train(olr, 100);
Gson gson = ModelSerializer.gson();
String s = gson.toJson(olr);
OnlineLogisticRegression olr2 = gson.fromJson(new StringReader(s), OnlineLogisticRegression.class);
+ OnlineLogisticRegression olr3 = roundTrip(olr, OnlineLogisticRegression.class);
assertEquals(0, olr.getBeta().minus(olr2.getBeta()).aggregate(Functions.MAX, Functions.IDENTITY), 1.0e-6);
+ assertEquals(0, olr.getBeta().minus(olr3.getBeta()).aggregate(Functions.MAX, Functions.IDENTITY), 1.0e-6);
train(olr, 100);
train(olr2, 100);
+ train(olr3, 100);
assertEquals(0, olr.getBeta().minus(olr2.getBeta()).aggregate(Functions.MAX, Functions.IDENTITY), 1.0e-6);
+ assertEquals(0, olr.getBeta().minus(olr3.getBeta()).aggregate(Functions.MAX, Functions.IDENTITY), 1.0e-6);
}
@Test
- public void crossFoldLearnerRoundTrip() {
+ public void crossFoldLearnerRoundTrip() throws IOException {
CrossFoldLearner learner = new CrossFoldLearner(5, 2, 5, new L1());
train(learner, 100);
Gson gson = ModelSerializer.gson();
String s = gson.toJson(learner);
CrossFoldLearner olr2 = gson.fromJson(new StringReader(s), CrossFoldLearner.class);
+ CrossFoldLearner olr3 = roundTrip(olr2, CrossFoldLearner.class);
double auc1 = learner.auc();
assertTrue(auc1 > 0.85);
assertEquals(auc1, olr2.auc(), 1.0e-6);
+ assertEquals(auc1, olr3.auc(), 1.0e-6);
train(learner, 100);
train(olr2, 100);
+ train(olr3, 100);
assertEquals(learner.auc(), olr2.auc(), 0.02);
+ assertEquals(learner.auc(), olr3.auc(), 0.02);
double auc2 = learner.auc();
assertTrue(auc2 > auc1);
}
@Test
- public void adaptiveLogisticRegressionRoundTrip() {
+ public void adaptiveLogisticRegressionRoundTrip() throws IOException {
AdaptiveLogisticRegression learner = new AdaptiveLogisticRegression(2, 5, new L1());
learner.setInterval(200);
train(learner, 400);
- Gson gson = ModelSerializer.gson();
- String s = gson.toJson(learner);
- AdaptiveLogisticRegression olr2 = gson.fromJson(new StringReader(s), AdaptiveLogisticRegression.class);
+ String s = ModelSerializer.gson().toJson(learner);
+ AdaptiveLogisticRegression olr2 = ModelSerializer.gson().fromJson(new StringReader(s), AdaptiveLogisticRegression.class);
+ AdaptiveLogisticRegression olr3 = roundTrip(learner, AdaptiveLogisticRegression.class);
double auc1 = learner.auc();
assertTrue(auc1 > 0.85);
assertEquals(auc1, olr2.auc(), 1.0e-6);
+ assertEquals(auc1, olr3.auc(), 1.0e-6);
train(learner, 1000);
train(olr2, 1000);
+ train(olr3, 1000);
assertEquals(learner.auc(), olr2.auc(), 0.005);
+ assertEquals(learner.auc(), olr3.auc(), 0.005);
double auc2 = learner.auc();
assertTrue(String.format("%.3f > %.3f", auc2, auc1), auc2 > auc1);
}