You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by df...@apache.org on 2013/06/02 20:56:27 UTC
svn commit: r1488766 - in /mahout/trunk: ./
core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/
core/src/test/java/org/apache/mahout/clustering/streaming/mapreduce/
Author: dfilimon
Date: Sun Jun 2 18:56:27 2013
New Revision: 1488766
URL: http://svn.apache.org/r1488766
Log:
MAHOUT-1224: Add the option of running a StreamingKMeans pass in the Reducer
Modified:
mahout/trunk/CHANGELOG
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansReducer.java
mahout/trunk/core/src/test/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansTestMR.java
Modified: mahout/trunk/CHANGELOG
URL: http://svn.apache.org/viewvc/mahout/trunk/CHANGELOG?rev=1488766&r1=1488765&r2=1488766&view=diff
==============================================================================
--- mahout/trunk/CHANGELOG (original)
+++ mahout/trunk/CHANGELOG Sun Jun 2 18:56:27 2013
@@ -2,6 +2,8 @@ Mahout Change Log
Release 0.8 - unreleased
+__MAHOUT-1224: Add the option of running a StreamingKMeans pass in the Reducer before BallKMeans (dfilimon)
+
MAHOUT-993: Some vector dumper flags are expecting arguments. (Andrew Look via robinanil)
MAHOUT-1228: Cleanup .gitignore (Stevo Slavic via ssc)
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansDriver.java?rev=1488766&r1=1488765&r2=1488766&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansDriver.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansDriver.java Sun Jun 2 18:56:27 2013
@@ -139,6 +139,12 @@ public final class StreamingKMeansDriver
*/
public static final String SEARCH_SIZE_OPTION = "searchSize";
+ /**
+ * Whether to run another pass of StreamingKMeans on the reducer's points before BallKMeans. On some data sets
+ * with a large number of mappers, the intermediate number of
+ */
+ public static final String REDUCE_STREAMING_KMEANS = "reduceStreamingKMeans";
+
private static final Logger log = LoggerFactory.getLogger(StreamingKMeansDriver.class);
private static final double INVALID_DISTANCE_CUTOFF = -1;
@@ -215,6 +221,10 @@ public final class StreamingKMeansDriver
"elements whose distances from the query vector is actually computer is proportional to " +
"searchSize. If no value is given, defaults to 1.", String.valueOf(2));
+ addFlag(REDUCE_STREAMING_KMEANS, "rskm", "There might be too many intermediate clusters from the mapper " +
+ "to fit into memory, so the reducer can run another pass of StreamingKMeans to collapse them down to a " +
+ "fewer clusters");
+
addOption(DefaultOptionCreator.methodOption().create());
if (parseArguments(args) == null) {
@@ -277,6 +287,8 @@ public final class StreamingKMeansDriver
numProjections = Integer.parseInt(getOption(NUM_PROJECTIONS_OPTION));
}
+ boolean reduceStreamingKMeans = hasOption(REDUCE_STREAMING_KMEANS);
+
configureOptionsForWorkers(getConf(), numClusters,
/* StreamingKMeans */
estimatedNumMapClusters, estimatedDistanceCutoff,
@@ -284,7 +296,8 @@ public final class StreamingKMeansDriver
maxNumIterations, trimFraction, randomInit, ignoreWeights, testProbability, numBallKMeansRuns,
/* Searcher */
measureClass, searcherClass, searchSize, numProjections,
- method);
+ method,
+ reduceStreamingKMeans);
}
/**
@@ -319,7 +332,8 @@ public final class StreamingKMeansDriver
/* Searcher */
String measureClass, String searcherClass,
int searchSize, int numProjections,
- String method) throws ClassNotFoundException {
+ String method,
+ boolean reduceStreamingKMeans) throws ClassNotFoundException {
// Checking preconditions for the parameters.
Preconditions.checkArgument(numClusters > 0, "Invalid number of clusters requested");
@@ -369,14 +383,17 @@ public final class StreamingKMeansDriver
conf.setInt(SEARCH_SIZE_OPTION, searchSize);
conf.setInt(NUM_PROJECTIONS_OPTION, numProjections);
conf.set(DefaultOptionCreator.METHOD_OPTION, method);
+
+ conf.setBoolean(REDUCE_STREAMING_KMEANS, reduceStreamingKMeans);
+
log.info("Parameters are: [k] numClusters {}; " +
"[SKM] estimatedNumMapClusters {}; estimatedDistanceCutoff {} " +
"[BKM] maxNumIterations {}; trimFraction {}; randomInit {}; ignoreWeights {}; " +
"testProbability {}; numBallKMeansRuns {}; " +
"[S] measureClass {}; searcherClass {}; searcherSize {}; numProjections {}; " +
- "method {}", numClusters, estimatedNumMapClusters, estimatedDistanceCutoff,
+ "method {}; reduceStreamingKMeans {}", numClusters, estimatedNumMapClusters, estimatedDistanceCutoff,
maxNumIterations, trimFraction, randomInit, ignoreWeights, testProbability, numBallKMeansRuns,
- measureClass, searcherClass, searchSize, numProjections, method);
+ measureClass, searcherClass, searchSize, numProjections, method, reduceStreamingKMeans);
}
/**
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansReducer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansReducer.java?rev=1488766&r1=1488765&r2=1488766&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansReducer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansReducer.java Sun Jun 2 18:56:27 2013
@@ -28,9 +28,11 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.mahout.clustering.streaming.cluster.BallKMeans;
+import org.apache.mahout.clustering.streaming.cluster.StreamingKMeans;
import org.apache.mahout.common.commandline.DefaultOptionCreator;
import org.apache.mahout.math.Centroid;
import org.apache.mahout.math.Vector;
+import org.apache.mahout.math.neighborhood.UpdatableSearcher;
public class StreamingKMeansReducer extends Reducer<IntWritable, CentroidWritable, IntWritable, CentroidWritable> {
/**
@@ -45,11 +47,38 @@ public class StreamingKMeansReducer exte
conf = context.getConfiguration();
}
+ private StreamingKMeans getStreamingKMeans(int numClusters) {
+ // At this point the configuration received from the Driver is assumed to be valid.
+ // No other checks are made.
+ UpdatableSearcher searcher = StreamingKMeansUtilsMR.searcherFromConfiguration(conf);
+ // There is no way of estimating the distance cutoff unless we have some data.
+ return new StreamingKMeans(searcher, numClusters,
+ conf.getFloat(StreamingKMeansDriver.ESTIMATED_DISTANCE_CUTOFF, 1e-4f));
+ }
+
@Override
public void reduce(IntWritable key, Iterable<CentroidWritable> centroids,
Context context) throws IOException, InterruptedException {
+ List<Centroid> intermediateCentroids;
+ // There might be too many intermediate centroids to fit into memory, in which case, we run another pass
+ // of StreamingKMeans to collapse the clusters further.
+ if (conf.getBoolean(StreamingKMeansDriver.REDUCE_STREAMING_KMEANS, false)) {
+ StreamingKMeans clusterer = getStreamingKMeans(conf.getInt(StreamingKMeansDriver.ESTIMATED_NUM_MAP_CLUSTERS, 1));
+ clusterer.cluster(Iterables.transform(centroids, new Function<CentroidWritable, Centroid>() {
+ @Override
+ public Centroid apply(org.apache.mahout.clustering.streaming.mapreduce.CentroidWritable input) {
+ Preconditions.checkNotNull(input);
+ return input.getCentroid();
+ }
+ }));
+ clusterer.reindexCentroids();
+ intermediateCentroids = Lists.newArrayList(clusterer);
+ } else {
+ intermediateCentroids = centroidWritablesToList(centroids);
+ }
+
int index = 0;
- for (Vector centroid : getBestCentroids(centroidWritablesToList(centroids), conf)) {
+ for (Vector centroid : getBestCentroids(intermediateCentroids, conf)) {
context.write(new IntWritable(index), new CentroidWritable((Centroid)centroid));
++index;
}
Modified: mahout/trunk/core/src/test/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansTestMR.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/test/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansTestMR.java?rev=1488766&r1=1488765&r2=1488766&view=diff
==============================================================================
--- mahout/trunk/core/src/test/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansTestMR.java (original)
+++ mahout/trunk/core/src/test/java/org/apache/mahout/clustering/streaming/mapreduce/StreamingKMeansTestMR.java Sun Jun 2 18:56:27 2013
@@ -84,6 +84,9 @@ public class StreamingKMeansTestMR {
(1 << NUM_DIMENSIONS) * (int)Math.log(NUM_DATA_POINTS));
configuration.setFloat(StreamingKMeansDriver.ESTIMATED_DISTANCE_CUTOFF, (float) DISTANCE_CUTOFF);
configuration.setInt(StreamingKMeansDriver.MAX_NUM_ITERATIONS, MAX_NUM_ITERATIONS);
+
+ // Collapse the Centroids in the reducer.
+ configuration.setBoolean(StreamingKMeansDriver.REDUCE_STREAMING_KMEANS, true);
}
@Parameterized.Parameters