You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by je...@apache.org on 2010/07/22 03:28:43 UTC
svn commit: r966474 [1/2] - in /mahout/trunk:
core/src/main/java/org/apache/mahout/clustering/canopy/
core/src/main/java/org/apache/mahout/clustering/dirichlet/
core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/
core/src/main/java/org/apache/...
Author: jeastman
Date: Thu Jul 22 01:28:42 2010
New Revision: 966474
URL: http://svn.apache.org/viewvc?rev=966474&view=rev
Log:
MAHOUT-294:
- made clustering driver intermediate job steps public where feasible
- added method=sequential|mapreduce argument to Canopy and Dirichlet to allow reference implementations to
be exercised from the command line
- more clustering drivers will follow
- added/updated unit tests to verify sequential implementations from command line
- all tests run
Modified:
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyClusterer.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyMapper.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyReducer.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterMapper.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterer.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletMapper.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletReducer.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansReducer.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/common/commandline/DefaultOptionCreator.java
mahout/trunk/core/src/test/java/org/apache/mahout/clustering/canopy/TestCanopyCreation.java
mahout/trunk/core/src/test/java/org/apache/mahout/clustering/dirichlet/TestMapReduce.java
mahout/trunk/core/src/test/java/org/apache/mahout/clustering/kmeans/TestKmeansClustering.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/canopy/Job.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/dirichlet/Job.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/fuzzykmeans/Job.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/kmeans/Job.java
mahout/trunk/utils/src/test/java/org/apache/mahout/clustering/TestClusterDumper.java
mahout/trunk/utils/src/test/java/org/apache/mahout/clustering/cdbw/TestCDbwEvaluator.java
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyClusterer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyClusterer.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyClusterer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyClusterer.java Thu Jul 22 01:28:42 2010
@@ -25,17 +25,23 @@ import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.mahout.clustering.ClusterBase;
import org.apache.mahout.clustering.WeightedVectorWritable;
import org.apache.mahout.common.distance.DistanceMeasure;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * @author jeff
+ *
+ */
public class CanopyClusterer {
- private int nextCanopyId;
+ private static final Logger log = LoggerFactory.getLogger(CanopyClusterer.class);
- private int numVectors;
+ private int nextCanopyId;
// the T1 distance threshold
private double t1;
@@ -94,37 +100,49 @@ public class CanopyClusterer {
* points. Because of this it does not need to actually store the points, instead storing a total points
* vector and the number of points. From this a centroid can be computed.
* <p/>
- * This method is used by the CanopyReducer.
+ * This method is used by the CanopyMapper, CanopyReducer and CanopyDriver.
*
* @param point
* the point to be added
* @param canopies
* the List<Canopy> to be appended
- * @param context
- * Object to report status to the MR interface
* @throws IOException
*/
- public void addPointToCanopies(Vector point, List<Canopy> canopies, TaskAttemptContext context) throws IOException {
+ public void addPointToCanopies(Vector point, List<Canopy> canopies) throws IOException {
boolean pointStronglyBound = false;
for (Canopy canopy : canopies) {
double dist = measure.distance(canopy.getCenter().getLengthSquared(), canopy.getCenter(), point);
if (dist < t1) {
+ log.info("Added point: " + ClusterBase.formatVector(point, null) + " to canopy: " + canopy.getIdentifier());
canopy.addPoint(point);
}
pointStronglyBound = pointStronglyBound || (dist < t2);
}
if (!pointStronglyBound) {
- context.setStatus("Created new Canopy:" + nextCanopyId + " numPoints:" + numVectors);
+ log.info("Created new Canopy:" + nextCanopyId + " at center:" + ClusterBase.formatVector(point, null));
canopies.add(new Canopy(point, nextCanopyId++));
}
- numVectors++;
}
+ /**
+ * Emit the point to the closest Canopy
+ *
+ * @param point
+ * @param canopies
+ * @param context
+ * @throws IOException
+ * @throws InterruptedException
+ */
public void emitPointToClosestCanopy(Vector point,
List<Canopy> canopies,
- Mapper<?,?,IntWritable,WeightedVectorWritable>.Context context)
- throws IOException, InterruptedException {
+ Mapper<?, ?, IntWritable, WeightedVectorWritable>.Context context) throws IOException,
+ InterruptedException {
+ Canopy closest = findClosestCanopy(point, canopies);
+ context.write(new IntWritable(closest.getId()), new WeightedVectorWritable(1, new VectorWritable(point)));
+ context.setStatus("Emit Closest Canopy ID:" + closest.getIdentifier());
+ }
+ protected Canopy findClosestCanopy(Vector point, List<Canopy> canopies) {
double minDist = Double.MAX_VALUE;
Canopy closest = null;
// find closest canopy
@@ -135,9 +153,7 @@ public class CanopyClusterer {
closest = canopy;
}
}
- // emit to closest canopy
- context.write(new IntWritable(closest.getId()), new WeightedVectorWritable(1, new VectorWritable(point)));
- context.setStatus("Emit Closest Canopy ID:" + closest.getIdentifier());
+ return closest;
}
/**
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyDriver.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyDriver.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyDriver.java Thu Jul 22 01:28:42 2010
@@ -18,22 +18,31 @@
package org.apache.mahout.clustering.canopy;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.mahout.clustering.Cluster;
+import org.apache.mahout.clustering.ClusterBase;
import org.apache.mahout.clustering.WeightedVectorWritable;
+import org.apache.mahout.clustering.kmeans.OutputLogFilter;
import org.apache.mahout.common.AbstractJob;
import org.apache.mahout.common.HadoopUtil;
import org.apache.mahout.common.commandline.DefaultOptionCreator;
+import org.apache.mahout.common.distance.DistanceMeasure;
import org.apache.mahout.math.VectorWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,7 +61,7 @@ public class CanopyDriver extends Abstra
}
/**
- * Run the job
+ * Run the job on a new driver instance (convenience)
*
* @param input
* the input pathname String
@@ -66,12 +75,21 @@ public class CanopyDriver extends Abstra
* the T2 distance threshold
* @param runClustering
* true if points are to be clustered after clusters are determined
+ * @param runSequential execute sequentially if true
* @throws ClassNotFoundException
* @throws InterruptedException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
*/
- public static void runJob(Path input, Path output, String measureClassName, double t1, double t2, boolean runClustering)
- throws IOException, InterruptedException, ClassNotFoundException {
- new CanopyDriver().job(input, output, measureClassName, t1, t2, runClustering);
+ public static void runJob(Path input,
+ Path output,
+ String measureClassName,
+ double t1,
+ double t2,
+ boolean runClustering,
+ boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException,
+ InstantiationException, IllegalAccessException {
+ new CanopyDriver().job(input, output, measureClassName, t1, t2, runClustering, runSequential);
}
@Override
@@ -84,6 +102,7 @@ public class CanopyDriver extends Abstra
addOption(DefaultOptionCreator.t2Option().create());
addOption(DefaultOptionCreator.overwriteOption().create());
addOption(DefaultOptionCreator.clusteringOption().create());
+ addOption(DefaultOptionCreator.methodOption().create());
Map<String, String> argMap = parseArguments(args);
if (argMap == null) {
@@ -99,14 +118,142 @@ public class CanopyDriver extends Abstra
double t1 = Double.parseDouble(getOption(DefaultOptionCreator.T1_OPTION));
double t2 = Double.parseDouble(getOption(DefaultOptionCreator.T2_OPTION));
boolean runClustering = hasOption(DefaultOptionCreator.CLUSTERING_OPTION);
+ boolean runSequential = (getOption(DefaultOptionCreator.METHOD_OPTION).equalsIgnoreCase(DefaultOptionCreator.SEQUENTIAL_METHOD));
- job(input, output, measureClass, t1, t2, runClustering);
+ job(input, output, measureClass, t1, t2, runClustering, runSequential);
return 0;
}
- private void job(Path input, Path output, String measureClassName, double t1, double t2, boolean runClustering)
- throws IOException, InterruptedException, ClassNotFoundException {
+ /**
+ * Build a directory of Canopy clusters from the input arguments and, if requested,
+ * cluster the input vectors using these clusters
+ * @param input the Path to the directory containing input vectors
+ * @param output the Path for all output directories
+ * @param measureClassName the String class name of the DistanceMeasure
+ * @param t1 the double T1 distance metric
+ * @param t2 the double T2 distance metric
+ * @param runClustering cluster the input vectors if true
+ * @param runSequential execute sequentially if true
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
+ */
+ public void job(Path input,
+ Path output,
+ String measureClassName,
+ double t1,
+ double t2,
+ boolean runClustering,
+ boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException, InstantiationException,
+ IllegalAccessException {
+ Path clustersOut = buildClusters(input, output, measureClassName, t1, t2, runSequential);
+ if (runClustering) {
+ clusterData(input, clustersOut, output, measureClassName, t1, t2, runSequential);
+ }
+ }
+
+ /**
+ * Build a directory of Canopy clusters from the input vectors and other arguments.
+ * Run sequential or mapreduce execution as requested
+ *
+ * @param input the Path to the directory containing input vectors
+ * @param output the Path for all output directories
+ * @param measureClassName the String class name of the DistanceMeasure
+ * @param t1 the double T1 distance metric
+ * @param t2 the double T2 distance metric
+ * @param runSequential a boolean indicates to run the sequential (reference) algorithm
+ * @return the canopy output directory Path
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
+ */
+ public Path buildClusters(Path input, Path output, String measureClassName, double t1, double t2, boolean runSequential)
+ throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
log.info("Input: {} Out: {} " + "Measure: {} t1: {} t2: {}", new Object[] { input, output, measureClassName, t1, t2 });
+ if (runSequential) {
+ return buildClustersSeq(input, output, measureClassName, t1, t2);
+ } else {
+ return buildClustersMR(input, output, measureClassName, t1, t2);
+ }
+ }
+
+ /**
+ * Build a directory of Canopy clusters from the input vectors and other arguments.
+ * Run sequential execution
+ *
+ * @param input the Path to the directory containing input vectors
+ * @param output the Path for all output directories
+ * @param measureClassName the String class name of the DistanceMeasure
+ * @param t1 the double T1 distance metric
+ * @param t2 the double T2 distance metric
+ * @param runSequential a boolean indicates to run sequential (reference) algorithm
+ * @return the canopy output directory Path
+ * @throws ClassNotFoundException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ * @throws IOException
+ */
+ private Path buildClustersSeq(Path input, Path output, String measureClassName, double t1, double t2)
+ throws InstantiationException, IllegalAccessException, ClassNotFoundException, IOException {
+ ClassLoader ccl = Thread.currentThread().getContextClassLoader();
+ DistanceMeasure measure = (DistanceMeasure) ccl.loadClass(measureClassName).newInstance();
+ CanopyClusterer clusterer = new CanopyClusterer(measure, t1, t2);
+ List<Canopy> canopies = new ArrayList<Canopy>();
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(input.toUri(), conf);
+ FileStatus[] status = fs.listStatus(input, new OutputLogFilter());
+ for (FileStatus s : status) {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
+ try {
+ WritableComparable<?> key = (WritableComparable<?>) reader.getKeyClass().newInstance();
+ VectorWritable vw = (VectorWritable) reader.getValueClass().newInstance();
+ while (reader.next(key, vw)) {
+ clusterer.addPointToCanopies(vw.get(), canopies);
+ vw = (VectorWritable) reader.getValueClass().newInstance();
+ }
+ } finally {
+ reader.close();
+ }
+ }
+ Path canopyOutputDir = new Path(output, Cluster.CLUSTERS_DIR + '0');
+ Path path = new Path(canopyOutputDir, "part-r-00000");
+ SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, path, Text.class, Canopy.class);
+ try {
+ for (Canopy canopy : canopies) {
+ log.info("Writing Canopy:" + canopy.getIdentifier() + " center:" + ClusterBase.formatVector(canopy.getCenter(), null)
+ + " numPoints:" + canopy.getNumPoints() + " centroid:" + ClusterBase.formatVector(canopy.computeCentroid(), null));
+ writer.append(new Text(canopy.getIdentifier()), canopy);
+ }
+ } finally {
+ writer.close();
+ }
+ return canopyOutputDir;
+ }
+
+ /**
+ * Build a directory of Canopy clusters from the input vectors and other arguments.
+ * Run mapreduce execution
+ *
+ * @param input the Path to the directory containing input vectors
+ * @param output the Path for all output directories
+ * @param measureClassName the String class name of the DistanceMeasure
+ * @param t1 the double T1 distance metric
+ * @param t2 the double T2 distance metric
+ * @param runSequential a boolean indicates to run sequential (reference) algorithm
+ * @return the canopy output directory Path
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ */
+ private Path buildClustersMR(Path input, Path output, String measureClassName, double t1, double t2) throws IOException,
+ InterruptedException, ClassNotFoundException {
Configuration conf = new Configuration();
conf.set(CanopyConfigKeys.DISTANCE_MEASURE_KEY, measureClassName);
conf.set(CanopyConfigKeys.T1_KEY, String.valueOf(t1));
@@ -128,33 +275,102 @@ public class CanopyDriver extends Abstra
Path canopyOutputDir = new Path(output, Cluster.CLUSTERS_DIR + '0');
FileOutputFormat.setOutputPath(job, canopyOutputDir);
job.waitForCompletion(true);
+ return canopyOutputDir;
+ }
- if (runClustering) {
- runClustering(input, canopyOutputDir, output, measureClassName, t1, t2);
+ /**
+ * Cluster the points using the given canopies and other arguments
+ *
+ * @param points the input points directory pathname String
+ * @param canopies the input canopies directory pathname String
+ * @param output the output directory pathname String
+ * @param measureClassName the DistanceMeasure class name
+ * @param t1 the T1 distance threshold
+ * @param t2 the T2 distance threshold
+ * @param runSequential a boolean indicates to run the sequential (reference) algorithm
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
+ */
+ public void clusterData(Path points,
+ Path canopies,
+ Path output,
+ String measureClassName,
+ double t1,
+ double t2,
+ boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException,
+ InstantiationException, IllegalAccessException {
+ if (runSequential) {
+ clusterDataSeq(points, canopies, output, measureClassName, t1, t2);
+ } else {
+ clusterDataMR(points, canopies, output, measureClassName, t1, t2);
+ }
+ }
+
+ private void clusterDataSeq(Path points, Path canopies, Path output, String measureClassName, double t1, double t2)
+ throws InstantiationException, IllegalAccessException, ClassNotFoundException, IOException, InterruptedException {
+ ClassLoader ccl = Thread.currentThread().getContextClassLoader();
+ DistanceMeasure measure = (DistanceMeasure) ccl.loadClass(measureClassName).newInstance();
+ CanopyClusterer clusterer = new CanopyClusterer(measure, t1, t2);
+
+ List<Canopy> clusters = new ArrayList<Canopy>();
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(canopies.toUri(), conf);
+ FileStatus[] status = fs.listStatus(canopies, new OutputLogFilter());
+ for (FileStatus s : status) {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
+ try {
+ Text key = (Text) reader.getKeyClass().newInstance();
+ Canopy value = (Canopy) reader.getValueClass().newInstance();
+ while (reader.next(key, value)) {
+ clusters.add(value);
+ value = (Canopy) reader.getValueClass().newInstance();
+ }
+ } finally {
+ reader.close();
+ }
+ }
+ // iterate over all points, assigning each to the closest canopy and outputing that clustering
+ fs = FileSystem.get(points.toUri(), conf);
+ status = fs.listStatus(points, new OutputLogFilter());
+ Path outPath = new Path(output, DEFAULT_CLUSTERED_POINTS_DIRECTORY);
+ int part = 0;
+ for (FileStatus s : status) {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
+ SequenceFile.Writer writer = new SequenceFile.Writer(fs,
+ conf,
+ new Path(outPath, "part-m-" + part++),
+ IntWritable.class,
+ WeightedVectorWritable.class);
+ try {
+ WritableComparable<?> key = (WritableComparable<?>) reader.getKeyClass().newInstance();
+ VectorWritable vw = (VectorWritable) reader.getValueClass().newInstance();
+ while (reader.next(key, vw)) {
+ Canopy closest = clusterer.findClosestCanopy(vw.get(), clusters);
+ writer.append(new IntWritable(closest.getId()), new WeightedVectorWritable(1, vw));
+ vw = (VectorWritable) reader.getValueClass().newInstance();
+ }
+ } finally {
+ reader.close();
+ writer.close();
+ }
}
}
/**
- * Run the job
- *
* @param points
- * the input points directory pathname String
* @param canopies
- * the input canopies directory pathname String
* @param output
- * the output directory pathname String
* @param measureClassName
- * the DistanceMeasure class name
* @param t1
- * the T1 distance threshold
* @param t2
- * the T2 distance threshold
- * @throws ClassNotFoundException
- * @throws InterruptedException
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
*/
- private void runClustering(Path points, Path canopies, Path output, String measureClassName, double t1, double t2)
+ private void clusterDataMR(Path points, Path canopies, Path output, String measureClassName, double t1, double t2)
throws IOException, InterruptedException, ClassNotFoundException {
-
Configuration conf = new Configuration();
conf.set(CanopyConfigKeys.DISTANCE_MEASURE_KEY, measureClassName);
conf.set(CanopyConfigKeys.T1_KEY, String.valueOf(t1));
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyMapper.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyMapper.java Thu Jul 22 01:28:42 2010
@@ -35,7 +35,7 @@ class CanopyMapper extends Mapper<Writab
@Override
protected void map(WritableComparable<?> key, VectorWritable point, Context context) throws IOException, InterruptedException {
- canopyClusterer.addPointToCanopies(point.get(), canopies, context);
+ canopyClusterer.addPointToCanopies(point.get(), canopies);
}
@Override
@@ -48,8 +48,7 @@ class CanopyMapper extends Mapper<Writab
protected void cleanup(Context context) throws IOException, InterruptedException {
for (Canopy canopy : canopies) {
Vector centroid = canopy.computeCentroid();
- VectorWritable vw = new VectorWritable(centroid);
- context.write(new Text("centroid"), vw);
+ context.write(new Text("centroid"), new VectorWritable(centroid));
}
super.cleanup(context);
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyReducer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyReducer.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyReducer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/CanopyReducer.java Thu Jul 22 01:28:42 2010
@@ -32,7 +32,7 @@ public class CanopyReducer extends Reduc
protected void reduce(Text arg0, Iterable<VectorWritable> values, Context context) throws IOException, InterruptedException {
for (VectorWritable value : values) {
Vector point = value.get();
- canopyClusterer.addPointToCanopies(point, canopies, context);
+ canopyClusterer.addPointToCanopies(point, canopies);
}
for (Canopy canopy : canopies) {
context.write(new Text(canopy.getIdentifier()), canopy);
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterMapper.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterMapper.java Thu Jul 22 01:28:42 2010
@@ -37,6 +37,7 @@ import org.apache.mahout.math.VectorWrit
public class DirichletClusterMapper extends Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable> {
private List<DirichletCluster<VectorWritable>> clusters;
+
private DirichletClusterer<VectorWritable> clusterer;
@Override
@@ -52,8 +53,7 @@ public class DirichletClusterMapper exte
clusters = getClusters(conf);
String emitMostLikely = conf.get(DirichletDriver.EMIT_MOST_LIKELY_KEY);
String threshold = conf.get(DirichletDriver.THRESHOLD_KEY);
- clusterer = new DirichletClusterer<VectorWritable>(Boolean.parseBoolean(emitMostLikely),
- Double.parseDouble(threshold));
+ clusterer = new DirichletClusterer<VectorWritable>(Boolean.parseBoolean(emitMostLikely), Double.parseDouble(threshold));
} catch (SecurityException e) {
throw new IllegalStateException(e);
} catch (IllegalArgumentException e) {
@@ -61,15 +61,18 @@ public class DirichletClusterMapper exte
}
}
- public static List<DirichletCluster<VectorWritable>> getClusters(Configuration job) {
- String statePath = job.get(DirichletDriver.STATE_IN_KEY);
+ public static List<DirichletCluster<VectorWritable>> getClusters(Configuration conf) {
+ String statePath = conf.get(DirichletDriver.STATE_IN_KEY);
+ return loadClusters(conf, new Path(statePath));
+ }
+
+ protected static List<DirichletCluster<VectorWritable>> loadClusters(Configuration conf, Path stateIn) {
List<DirichletCluster<VectorWritable>> clusters = new ArrayList<DirichletCluster<VectorWritable>>();
try {
- Path path = new Path(statePath);
- FileSystem fs = FileSystem.get(path.toUri(), job);
- FileStatus[] status = fs.listStatus(path, new OutputLogFilter());
+ FileSystem fs = FileSystem.get(stateIn.toUri(), conf);
+ FileStatus[] status = fs.listStatus(stateIn, new OutputLogFilter());
for (FileStatus s : status) {
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), job);
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
try {
Text key = new Text();
DirichletCluster<VectorWritable> cluster = new DirichletCluster<VectorWritable>();
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterer.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletClusterer.java Thu Jul 22 01:28:42 2010
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.clustering.WeightedVectorWritable;
import org.apache.mahout.clustering.dirichlet.models.Model;
@@ -111,6 +112,36 @@ public class DirichletClusterer<O> {
/**
* Create a new instance on the sample data with the given additional parameters
*
+ * @param points
+ * the observed data to be clustered
+ * @param modelFactory
+ * the ModelDistribution to use
+ * @param alpha0
+ * the double value for the beta distributions
+ * @param numClusters
+ * the int number of clusters
+ * @param thin
+ * the int thinning interval, used to report every n iterations
+ * @param burnin
+ * the int burnin interval, used to suppress early iterations
+ * @param numIterations
+ * number of iterations to be performed
+ */
+ public static List<Model<Vector>[]> clusterPoints(List<Vector> points,
+ ModelDistribution<Vector> modelFactory,
+ double alpha0,
+ int numClusters,
+ int thin,
+ int burnin,
+ int numIterations) {
+ DirichletClusterer<Vector> clusterer = new DirichletClusterer<Vector>(points, modelFactory, alpha0, numClusters, thin, burnin);
+ return clusterer.cluster(numIterations);
+
+ }
+
+ /**
+ * Create a new instance on the sample data with the given additional parameters
+ *
* @param sampleData
* the observed data to be clustered
* @param modelFactory
@@ -124,8 +155,12 @@ public class DirichletClusterer<O> {
* @param burnin
* the int burnin interval, used to suppress early iterations
*/
- public DirichletClusterer(List<O> sampleData, ModelDistribution<O> modelFactory, double alpha0, int numClusters, int thin,
- int burnin) {
+ public DirichletClusterer(List<O> sampleData,
+ ModelDistribution<O> modelFactory,
+ double alpha0,
+ int numClusters,
+ int thin,
+ int burnin) {
this.sampleData = sampleData;
this.modelFactory = modelFactory;
this.thin = thin;
@@ -151,6 +186,19 @@ public class DirichletClusterer<O> {
}
/**
+ * This constructor is used by DirichletMapper and DirichletReducer for setting up their clusterer
+ * @param state
+ */
+ public DirichletClusterer(DirichletState<O> state) {
+ this.state = state;
+ this.modelFactory = state.getModelFactory();
+ this.sampleData = null;
+ this.numClusters = state.getNumClusters();
+ this.thin = 0;
+ this.burnin = 0;
+ }
+
+ /**
* Iterate over the sample data, obtaining cluster samples periodically and returning them.
*
* @param numIterations
@@ -159,7 +207,7 @@ public class DirichletClusterer<O> {
*/
public List<Model<O>[]> cluster(int numIterations) {
for (int iteration = 0; iteration < numIterations; iteration++) {
- iterate(iteration, state);
+ iterate(iteration);
}
return clusterSamples;
}
@@ -167,24 +215,15 @@ public class DirichletClusterer<O> {
/**
* Perform one iteration of the clustering process, iterating over the samples to build a new array of
* models, then updating the state for the next iteration
- *
- * @param state
- * the DirichletState<Observation> of this iteration
*/
- private void iterate(int iteration, DirichletState<O> state) {
+ private void iterate(int iteration) {
// create new posterior models
Model<O>[] newModels = modelFactory.sampleFromPosterior(state.getModels());
// iterate over the samples, assigning each to a model
- for (O x : sampleData) {
- // compute normalized vector of probabilities that x is described by each model
- Vector pi = normalizedProbabilities(state, x);
- // then pick one cluster by sampling a Multinomial distribution based upon them
- // see: http://en.wikipedia.org/wiki/Multinomial_distribution
- int k = UncommonDistributions.rMultinom(pi);
- // ask the selected model to observe the datum
- newModels[k].observe(x);
+ for (O observation : sampleData) {
+ observe(newModels, observation);
}
// periodically add models to the cluster samples after the burn-in period
@@ -196,6 +235,55 @@ public class DirichletClusterer<O> {
}
/**
+ * @param newModels
+ * @param observation
+ */
+ protected void observe(Model<O>[] newModels, O observation) {
+ int k = assignToModel(observation);
+ // ask the selected model to observe the datum
+ newModels[k].observe(observation);
+ }
+
+ /**
+ * Assign the observation to one of the models based upon probabilities
+ * @param observation
+ * @return the assigned model's index
+ */
+ protected int assignToModel(O observation) {
+ // compute normalized vector of probabilities that x is described by each model
+ Vector pi = normalizedProbabilities(state, observation);
+ // then pick one cluster by sampling a Multinomial distribution based upon them
+ // see: http://en.wikipedia.org/wiki/Multinomial_distribution
+ int k = UncommonDistributions.rMultinom(pi);
+ return k;
+ }
+
+ @SuppressWarnings("unchecked")
+ protected void updateModels(Model<VectorWritable>[] newModels) {
+ state.update((Model<O>[]) newModels);
+ }
+
+ /**
+ * @return
+ */
+ protected Model<O>[] samplePosteriorModels() {
+ return state.getModelFactory().sampleFromPosterior(state.getModels());
+ }
+
+ /**
+ * @param model
+ * @param k
+ * @return
+ */
+ @SuppressWarnings("unchecked")
+ protected DirichletCluster<VectorWritable> updateCluster(Model<VectorWritable> model, int k) {
+ model.computeParameters();
+ DirichletCluster<O> cluster = state.getClusters().get(k);
+ cluster.setModel((Model<O>) model);
+ return (DirichletCluster<VectorWritable>) cluster;
+ }
+
+ /**
* Compute a normalized vector of probabilities that x is described by each model using the mixture and the
* model pdfs
*
@@ -220,9 +308,19 @@ public class DirichletClusterer<O> {
return pi;
}
- public void emitPointToClusters(VectorWritable vector, List<DirichletCluster<VectorWritable>> clusters,
- Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable>.Context context) throws IOException,
- InterruptedException {
+ /**
+ * Emit the point to one or more clusters depending upon clusterer state
+ *
+ * @param vector a VectorWritable holding the Vector
+ * @param clusters a List of DirichletClusters
+ * @param context a Mapper.Context to emit to
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public void emitPointToClusters(VectorWritable vector,
+ List<DirichletCluster<VectorWritable>> clusters,
+ Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable>.Context context)
+ throws IOException, InterruptedException {
Vector pi = new DenseVector(clusters.size());
for (int i = 0; i < clusters.size(); i++) {
pi.set(i, clusters.get(i).getModel().pdf(vector));
@@ -237,11 +335,19 @@ public class DirichletClusterer<O> {
/**
* Emit the point to the most likely cluster
+ *
+ * @param vector a VectorWritable holding the Vector
+ * @param clusters a List of DirichletClusters
* @param pi the normalized pdf Vector for the point
+ * @param context a Mapper.Context to emit to
+ * @throws IOException
+ * @throws InterruptedException
*/
- private void emitMostLikelyCluster(VectorWritable point, List<DirichletCluster<VectorWritable>> clusters, Vector pi,
- Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable>.Context context) throws IOException,
- InterruptedException {
+ private void emitMostLikelyCluster(VectorWritable point,
+ List<DirichletCluster<VectorWritable>> clusters,
+ Vector pi,
+ Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable>.Context context)
+ throws IOException, InterruptedException {
int clusterId = -1;
double clusterPdf = 0;
for (int i = 0; i < clusters.size(); i++) {
@@ -255,9 +361,20 @@ public class DirichletClusterer<O> {
context.write(new IntWritable(clusterId), new WeightedVectorWritable(clusterPdf, point));
}
- private void emitAllClusters(VectorWritable point, List<DirichletCluster<VectorWritable>> clusters, Vector pi,
- Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable>.Context context) throws IOException,
- InterruptedException {
+ /**
+ * Emit the point to all clusters if pdf exceeds the threshold
+ * @param vector a VectorWritable holding the Vector
+ * @param clusters a List of DirichletClusters
+ * @param pi the normalized pdf Vector for the point
+ * @param context a Mapper.Context to emit to
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ private void emitAllClusters(VectorWritable point,
+ List<DirichletCluster<VectorWritable>> clusters,
+ Vector pi,
+ Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable>.Context context)
+ throws IOException, InterruptedException {
for (int i = 0; i < clusters.size(); i++) {
double pdf = pi.get(i);
if (pdf > threshold && clusters.get(i).getTotalCount() > 0) {
@@ -268,27 +385,66 @@ public class DirichletClusterer<O> {
}
/**
- * Create a new instance on the sample data with the given additional parameters
+ * Emit the point to one or more clusters depending upon clusterer state
*
- * @param points
- * the observed data to be clustered
- * @param modelFactory
- * the ModelDistribution to use
- * @param alpha0
- * the double value for the beta distributions
- * @param numClusters
- * the int number of clusters
- * @param thin
- * the int thinning interval, used to report every n iterations
- * @param burnin
- * the int burnin interval, used to suppress early iterations
- * @param numIterations
- * number of iterations to be performed
+ * @param vector a VectorWritable holding the Vector
+ * @param clusters a List of DirichletClusters
+ * @param writer a SequenceFile.Writer to emit to
+ * @throws IOException
*/
- public static List<Model<Vector>[]> clusterPoints(List<Vector> points, ModelDistribution<Vector> modelFactory, double alpha0,
- int numClusters, int thin, int burnin, int numIterations) {
- DirichletClusterer<Vector> clusterer = new DirichletClusterer<Vector>(points, modelFactory, alpha0, numClusters, thin, burnin);
- return clusterer.cluster(numIterations);
+ public void emitPointToClusters(VectorWritable vector, List<DirichletCluster<VectorWritable>> clusters, Writer writer)
+ throws IOException {
+ Vector pi = new DenseVector(clusters.size());
+ for (int i = 0; i < clusters.size(); i++) {
+ pi.set(i, clusters.get(i).getModel().pdf(vector));
+ }
+ pi = pi.divide(pi.zSum());
+ if (emitMostLikely) {
+ emitMostLikelyCluster(vector, clusters, pi, writer);
+ } else {
+ emitAllClusters(vector, clusters, pi, writer);
+ }
+ }
+ /**
+ * Emit the point to all clusters if pdf exceeds the threshold
+ *
+ * @param vector a VectorWritable holding the Vector
+ * @param clusters a List of DirichletClusters
+ * @param pi the normalized pdf Vector for the point
+ * @param writer a SequenceFile.Writer to emit to
+ * @throws IOException
+ */
+ private void emitAllClusters(VectorWritable vector, List<DirichletCluster<VectorWritable>> clusters, Vector pi, Writer writer)
+ throws IOException {
+ for (int i = 0; i < clusters.size(); i++) {
+ double pdf = pi.get(i);
+ if (pdf > threshold && clusters.get(i).getTotalCount() > 0) {
+ //System.out.println(i + ": " + ClusterBase.formatVector(vector.get(), null));
+ writer.append(new IntWritable(i), new WeightedVectorWritable(pdf, vector));
+ }
+ }
+ }
+
+ /**
+ * Emit the point to the most likely cluster
+ *
+ * @param vector a VectorWritable holding the Vector
+ * @param clusters a List of DirichletClusters
+ * @param pi the normalized pdf Vector for the point
+ * @param writer a SequenceFile.Writer to emit to
+ * @throws IOException
+ */
+ private void emitMostLikelyCluster(VectorWritable vector,
+ List<DirichletCluster<VectorWritable>> clusters,
+ Vector pi,
+ Writer writer) throws IOException {
+ for (int i = 0; i < clusters.size(); i++) {
+ double pdf = pi.get(i);
+ if (pdf > threshold && clusters.get(i).getTotalCount() > 0) {
+ //System.out.println(i + ": " + ClusterBase.formatVector(vector.get(), null));
+ writer.append(new IntWritable(i), new WeightedVectorWritable(pdf, vector));
+ }
+ }
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletDriver.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletDriver.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletDriver.java Thu Jul 22 01:28:42 2010
@@ -20,6 +20,7 @@ package org.apache.mahout.clustering.dir
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
+import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
@@ -37,6 +38,7 @@ import org.apache.hadoop.mapreduce.lib.o
import org.apache.mahout.clustering.Cluster;
import org.apache.mahout.clustering.WeightedVectorWritable;
import org.apache.mahout.clustering.dirichlet.models.AbstractVectorModelDistribution;
+import org.apache.mahout.clustering.dirichlet.models.Model;
import org.apache.mahout.clustering.dirichlet.models.NormalModelDistribution;
import org.apache.mahout.clustering.kmeans.OutputLogFilter;
import org.apache.mahout.common.AbstractJob;
@@ -74,7 +76,7 @@ public class DirichletDriver extends Abs
private static final Logger log = LoggerFactory.getLogger(DirichletDriver.class);
- protected DirichletDriver() {
+ public DirichletDriver() {
}
public static void main(String[] args) throws Exception {
@@ -104,6 +106,7 @@ public class DirichletDriver extends Abs
addOption(DefaultOptionCreator.emitMostLikelyOption().create());
addOption(DefaultOptionCreator.thresholdOption().create());
addOption(DefaultOptionCreator.numReducersOption().create());
+ addOption(DefaultOptionCreator.methodOption().create());
if (parseArguments(args) == null) {
return -1;
@@ -123,6 +126,7 @@ public class DirichletDriver extends Abs
double threshold = Double.parseDouble(getOption(DefaultOptionCreator.THRESHOLD_OPTION));
double alpha0 = Double.parseDouble(getOption(ALPHA_OPTION));
boolean runClustering = hasOption(DefaultOptionCreator.CLUSTERING_OPTION);
+ boolean runSequential = (getOption(DefaultOptionCreator.METHOD_OPTION).equalsIgnoreCase(DefaultOptionCreator.SEQUENTIAL_METHOD));
job(input,
output,
@@ -134,12 +138,13 @@ public class DirichletDriver extends Abs
numReducers,
runClustering,
emitMostLikely,
- threshold);
+ threshold,
+ runSequential);
return 0;
}
/**
- * Run the job using supplied arguments
+ * Run the job using supplied arguments on a new driver instance (convenience)
*
* @param input
* the directory pathname for input points
@@ -163,6 +168,7 @@ public class DirichletDriver extends Abs
* a boolean if true emit only most likely cluster for each point
* @param threshold
* a double threshold value emits all clusters having greater pdf (emitMostLikely = false)
+ * @param runSequential execute sequentially if true
* @throws InterruptedException
*/
public static void runJob(Path input,
@@ -175,7 +181,8 @@ public class DirichletDriver extends Abs
int numReducers,
boolean runClustering,
boolean emitMostLikely,
- double threshold) throws ClassNotFoundException, InstantiationException, IllegalAccessException,
+ double threshold,
+ boolean runSequential) throws ClassNotFoundException, InstantiationException, IllegalAccessException,
IOException, SecurityException, NoSuchMethodException, InvocationTargetException, InterruptedException {
new DirichletDriver().job(input,
@@ -188,7 +195,8 @@ public class DirichletDriver extends Abs
numReducers,
runClustering,
emitMostLikely,
- threshold);
+ threshold,
+ runSequential);
}
/**
@@ -254,7 +262,7 @@ public class DirichletDriver extends Abs
/**
* Write initial state (prior distribution) to the output path directory
* @param output the output Path
- * @param stateIn the state input Path
+ * @param stateOut the state output Path
* @param modelFactory the String class name of the modelFactory
* @param modelPrototype the String class name of the modelPrototype
* @param prototypeSize the int size of the modelPrototype vectors
@@ -269,7 +277,7 @@ public class DirichletDriver extends Abs
* @throws InvocationTargetException
*/
private void writeInitialState(Path output,
- Path stateIn,
+ Path stateOut,
String modelFactory,
String modelPrototype,
int prototypeSize,
@@ -278,10 +286,14 @@ public class DirichletDriver extends Abs
IOException, SecurityException, NoSuchMethodException, InvocationTargetException {
DirichletState<VectorWritable> state = createState(modelFactory, modelPrototype, prototypeSize, numModels, alpha0);
+ writeState(output, stateOut, numModels, state);
+ }
+
+ private void writeState(Path output, Path stateOut, int numModels, DirichletState<VectorWritable> state) throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(output.toUri(), conf);
for (int i = 0; i < numModels; i++) {
- Path path = new Path(stateIn, "part-" + i);
+ Path path = new Path(stateOut, "part-" + i);
SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, path, Text.class, DirichletCluster.class);
writer.append(new Text(Integer.toString(i)), state.getClusters().get(i));
writer.close();
@@ -350,67 +362,92 @@ public class DirichletDriver extends Abs
}
/**
- * Run the job using supplied arguments
+ * Iterate over the input vectors to produce clusters and, if requested, use the
+ * results of the final iteration to cluster the input vectors.
*
* @param input
- * the directory pathname for input points
- * @param stateIn
- * the directory pathname for input state
+ * the directory Path for input points
* @param output
- * the directory pathname for output points
+ * the directory Path for output points
+ * @param modelFactory
+ * the String ModelDistribution class name to use
+ * @param modelPrototype
+ * the String class name of the model's prototype vector
+ * @param numClusters
+ * the number of models to iterate over
+ * @param maxIterations
+ * the maximum number of iterations
+ * @param alpha0
+ * the alpha_0 value for the DirichletDistribution
+ * @param numReducers
+ * the number of Reducers desired
+ * @param runClustering
+ * true if clustering of points to be done after iterations
* @param emitMostLikely
* a boolean if true emit only most likely cluster for each point
* @param threshold
* a double threshold value emits all clusters having greater pdf (emitMostLikely = false)
- * @throws ClassNotFoundException
- * @throws InterruptedException
- * @throws IOException
+ * @param runSequential execute sequentially if true
+ * @throws IOException
+ * @throws InstantiationException
+ * @throws IllegalAccessException
+ * @throws ClassNotFoundException
+ * @throws NoSuchMethodException
+ * @throws InvocationTargetException
+ * @throws InterruptedException
*/
- private void runClustering(Path input, Path stateIn, Path output, boolean emitMostLikely, double threshold) throws IOException,
- InterruptedException, ClassNotFoundException {
- Configuration conf = new Configuration();
- conf.set(STATE_IN_KEY, stateIn.toString());
- conf.set(EMIT_MOST_LIKELY_KEY, Boolean.toString(emitMostLikely));
- conf.set(THRESHOLD_KEY, Double.toString(threshold));
- Job job = new Job(conf);
- job.setOutputKeyClass(IntWritable.class);
- job.setOutputValueClass(WeightedVectorWritable.class);
- job.setMapperClass(DirichletClusterMapper.class);
- job.setInputFormatClass(SequenceFileInputFormat.class);
- job.setOutputFormatClass(SequenceFileOutputFormat.class);
- job.setNumReduceTasks(0);
- job.setJarByClass(DirichletDriver.class);
-
- FileInputFormat.addInputPath(job, input);
- FileOutputFormat.setOutputPath(job, output);
-
- job.waitForCompletion(true);
+ public void job(Path input,
+ Path output,
+ String modelFactory,
+ String modelPrototype,
+ int numClusters,
+ int maxIterations,
+ double alpha0,
+ int numReducers,
+ boolean runClustering,
+ boolean emitMostLikely,
+ double threshold,
+ boolean runSequential) throws IOException, InstantiationException, IllegalAccessException,
+ ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InterruptedException {
+ Path clustersOut = buildClusters(input,
+ output,
+ modelFactory,
+ modelPrototype,
+ numClusters,
+ maxIterations,
+ alpha0,
+ numReducers,
+ runSequential);
+ if (runClustering) {
+ clusterData(input, clustersOut, new Path(output, Cluster.CLUSTERED_POINTS_DIR), emitMostLikely, threshold, runSequential);
+ }
}
/**
- * Run the job
+ * Iterate over the input vectors to produce cluster directories for each iteration
+ *
* @param input
- * the directory pathname for input points
+ * the directory Path for input points
* @param output
- * the directory pathname for output points
+ * the directory Path for output points
* @param modelFactory
* the String ModelDistribution class name to use
* @param modelPrototype
- * the String class name of the model prototype
+ * the String class name of the model's prototype vector
* @param numClusters
- * the number of models
+ * the number of models to iterate over
* @param maxIterations
* the maximum number of iterations
* @param alpha0
* the alpha_0 value for the DirichletDistribution
* @param numReducers
* the number of Reducers desired
- * @param runClustering
- * true if clustering of points to be done after iterations
+ * @param runSequential execute sequentially if true
* @param emitMostLikely
* a boolean if true emit only most likely cluster for each point
* @param threshold
* a double threshold value emits all clusters having greater pdf (emitMostLikely = false)
+ * @return the Path of the final clusters directory
* @throws IOException
* @throws InstantiationException
* @throws IllegalAccessException
@@ -419,24 +456,143 @@ public class DirichletDriver extends Abs
* @throws InvocationTargetException
* @throws InterruptedException
*/
- private void job(Path input,
- Path output,
- String modelFactory,
- String modelPrototype,
- int numClusters,
- int maxIterations,
- double alpha0,
- int numReducers,
- boolean runClustering,
- boolean emitMostLikely,
- double threshold) throws IOException, InstantiationException, IllegalAccessException, ClassNotFoundException,
- NoSuchMethodException, InvocationTargetException, InterruptedException {
+ public Path buildClusters(Path input,
+ Path output,
+ String modelFactory,
+ String modelPrototype,
+ int numClusters,
+ int maxIterations,
+ double alpha0,
+ int numReducers,
+ boolean runSequential) throws IOException, InstantiationException, IllegalAccessException,
+ ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InterruptedException {
Path clustersIn = new Path(output, Cluster.INITIAL_CLUSTERS_DIR);
int protoSize = readPrototypeSize(input);
writeInitialState(output, clustersIn, modelFactory, modelPrototype, protoSize, numClusters, alpha0);
+ if (runSequential) {
+ clustersIn = buildClustersSeq(input,
+ output,
+ modelFactory,
+ modelPrototype,
+ numClusters,
+ maxIterations,
+ alpha0,
+ numReducers,
+ clustersIn,
+ protoSize);
+ } else {
+ clustersIn = buildClustersMR(input,
+ output,
+ modelFactory,
+ modelPrototype,
+ numClusters,
+ maxIterations,
+ alpha0,
+ numReducers,
+ clustersIn,
+ protoSize);
+ }
+ return clustersIn;
+ }
+
+ /**
+ * @param input
+ * @param output
+ * @param modelFactory
+ * @param modelPrototype
+ * @param numClusters
+ * @param maxIterations
+ * @param alpha0
+ * @param numReducers
+ * @param clustersIn
+ * @param protoSize
+ * @return
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ * @throws InvocationTargetException
+ * @throws NoSuchMethodException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
+ */
+ private Path buildClustersSeq(Path input,
+ Path output,
+ String modelFactory,
+ String modelPrototype,
+ int numClusters,
+ int maxIterations,
+ double alpha0,
+ int numReducers,
+ Path clustersIn,
+ int protoSize) throws IOException, InterruptedException, ClassNotFoundException,
+ InstantiationException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+ for (int iteration = 1; iteration <= maxIterations; iteration++) {
+ log.info("Iteration {}", iteration);
+ // point the output to a new directory per iteration
+ Path clustersOut = new Path(output, Cluster.CLUSTERS_DIR + iteration);
+ DirichletState<VectorWritable> state = DirichletMapper.loadState(new Configuration(),
+ clustersIn.toString(),
+ modelFactory,
+ modelPrototype,
+ alpha0,
+ protoSize,
+ numClusters);
+ Model<VectorWritable>[] newModels = state.getModelFactory().sampleFromPosterior(state.getModels());
+ DirichletClusterer<VectorWritable> clusterer = new DirichletClusterer<VectorWritable>(state);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(input.toUri(), conf);
+ FileStatus[] status = fs.listStatus(input, new OutputLogFilter());
+ for (FileStatus s : status) {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
+ try {
+ WritableComparable<?> key = (WritableComparable<?>) reader.getKeyClass().newInstance();
+ VectorWritable vw = (VectorWritable) reader.getValueClass().newInstance();
+ while (reader.next(key, vw)) {
+ clusterer.observe(newModels, vw);
+ vw = (VectorWritable) reader.getValueClass().newInstance();
+ }
+ } finally {
+ reader.close();
+ }
+ }
+ clusterer.updateModels(newModels);
+ writeState(output, clustersOut, numClusters, state);
+
+ // now point the input to the old output directory
+ clustersIn = clustersOut;
+ }
+ return clustersIn;
+ }
+
+ /**
+ * @param input
+ * @param output
+ * @param modelFactory
+ * @param modelPrototype
+ * @param numClusters
+ * @param maxIterations
+ * @param alpha0
+ * @param numReducers
+ * @param clustersIn
+ * @param protoSize
+ * @return
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ */
+ private Path buildClustersMR(Path input,
+ Path output,
+ String modelFactory,
+ String modelPrototype,
+ int numClusters,
+ int maxIterations,
+ double alpha0,
+ int numReducers,
+ Path clustersIn,
+ int protoSize) throws IOException, InterruptedException, ClassNotFoundException {
for (int iteration = 1; iteration <= maxIterations; iteration++) {
log.info("Iteration {}", iteration);
// point the output to a new directory per iteration
@@ -445,9 +601,97 @@ public class DirichletDriver extends Abs
// now point the input to the old output directory
clustersIn = clustersOut;
}
- if (runClustering) {
- // now cluster the most likely points
- runClustering(input, clustersIn, new Path(output, Cluster.CLUSTERED_POINTS_DIR), emitMostLikely, threshold);
+ return clustersIn;
+ }
+
+ /**
+ * Run the job using supplied arguments
+ *
+ * @param input
+ * the directory pathname for input points
+ * @param stateIn
+ * the directory pathname for input state
+ * @param output
+ * the directory pathname for output points
+ * @param emitMostLikely
+ * a boolean if true emit only most likely cluster for each point
+ * @param threshold
+ * a double threshold value emits all clusters having greater pdf (emitMostLikely = false)
+ * @param runSequential execute sequentially if true
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
+ * @throws IOException
+ * @throws IllegalAccessException
+ * @throws InstantiationException
+ */
+ public void clusterData(Path input, Path stateIn, Path output, boolean emitMostLikely, double threshold, boolean runSequential)
+ throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
+ if (runSequential) {
+ clusterDataSeq(input, stateIn, output, emitMostLikely, threshold);
+ } else {
+ clusterDataMR(input, stateIn, output, emitMostLikely, threshold);
+ }
+ }
+
+ private void clusterDataSeq(Path input, Path stateIn, Path output, boolean emitMostLikely, double threshold) throws IOException,
+ InstantiationException, IllegalAccessException {
+ Configuration conf = new Configuration();
+ List<DirichletCluster<VectorWritable>> clusters = DirichletClusterMapper.loadClusters(conf, stateIn);
+ DirichletClusterer<VectorWritable> clusterer = new DirichletClusterer<VectorWritable>(emitMostLikely, threshold);
+ // iterate over all points, assigning each to the closest canopy and outputing that clustering
+ FileSystem fs = FileSystem.get(input.toUri(), conf);
+ FileStatus[] status = fs.listStatus(input, new OutputLogFilter());
+ int part = 0;
+ for (FileStatus s : status) {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
+ SequenceFile.Writer writer = new SequenceFile.Writer(fs,
+ conf,
+ new Path(output, "part-m-" + part++),
+ IntWritable.class,
+ WeightedVectorWritable.class);
+ try {
+ WritableComparable<?> key = (WritableComparable<?>) reader.getKeyClass().newInstance();
+ VectorWritable vw = (VectorWritable) reader.getValueClass().newInstance();
+ while (reader.next(key, vw)) {
+ clusterer.emitPointToClusters(vw, clusters, writer);
+ vw = (VectorWritable) reader.getValueClass().newInstance();
+ }
+ } finally {
+ reader.close();
+ writer.close();
+ }
}
+
+ }
+
+ /**
+ * @param input
+ * @param stateIn
+ * @param output
+ * @param emitMostLikely
+ * @param threshold
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ */
+ private void clusterDataMR(Path input, Path stateIn, Path output, boolean emitMostLikely, double threshold) throws IOException,
+ InterruptedException, ClassNotFoundException {
+ Configuration conf = new Configuration();
+ conf.set(STATE_IN_KEY, stateIn.toString());
+ conf.set(EMIT_MOST_LIKELY_KEY, Boolean.toString(emitMostLikely));
+ conf.set(THRESHOLD_KEY, Double.toString(threshold));
+ Job job = new Job(conf);
+ job.setOutputKeyClass(IntWritable.class);
+ job.setOutputValueClass(WeightedVectorWritable.class);
+ job.setMapperClass(DirichletClusterMapper.class);
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setNumReduceTasks(0);
+ job.setJarByClass(DirichletDriver.class);
+
+ FileInputFormat.addInputPath(job, input);
+ FileOutputFormat.setOutputPath(job, output);
+
+ job.waitForCompletion(true);
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletMapper.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletMapper.java Thu Jul 22 01:28:42 2010
@@ -30,21 +30,15 @@ import org.apache.hadoop.io.WritableComp
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.clustering.kmeans.OutputLogFilter;
import org.apache.mahout.math.DenseVector;
-import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
-import org.apache.mahout.math.function.TimesFunction;
-public class DirichletMapper extends Mapper<WritableComparable<?>,VectorWritable,Text,VectorWritable> {
-
- private DirichletState<VectorWritable> state;
+public class DirichletMapper extends Mapper<WritableComparable<?>, VectorWritable, Text, VectorWritable> {
+
+ private DirichletClusterer<VectorWritable> clusterer;
@Override
protected void map(WritableComparable<?> key, VectorWritable v, Context context) throws IOException, InterruptedException {
- // compute a normalized vector of probabilities that v is described by each model
- Vector pi = normalizedProbabilities(state, v);
- // then pick one model by sampling a Multinomial distribution based upon them
- // see: http://en.wikipedia.org/wiki/Multinomial_distribution
- int k = UncommonDistributions.rMultinom(pi);
+ int k = clusterer.assignToModel(v);
context.write(new Text(String.valueOf(k)), v);
}
@@ -52,7 +46,13 @@ public class DirichletMapper extends Map
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
try {
- state = getDirichletState(context.getConfiguration());
+ DirichletState<VectorWritable> dirichletState = getDirichletState(context.getConfiguration());
+ clusterer = new DirichletClusterer<VectorWritable>(dirichletState);
+ for (int i = 0; i < dirichletState.getNumClusters(); i++) {
+ // write an empty vector to each clusterId so that all will be seen by a reducer
+ // Reducers will ignore these points but every model will be processed by one of them
+ context.write(new Text(Integer.toString(i)), new VectorWritable(new DenseVector(0)));
+ }
} catch (NumberFormatException e) {
throw new IllegalStateException(e);
} catch (SecurityException e) {
@@ -67,42 +67,26 @@ public class DirichletMapper extends Map
}
public void setup(DirichletState<VectorWritable> state) {
- this.state = state;
+ this.clusterer = new DirichletClusterer<VectorWritable>(state);
}
-
- public static DirichletState<VectorWritable> getDirichletState(Configuration conf)
- throws NoSuchMethodException, InvocationTargetException {
+
+ public static DirichletState<VectorWritable> getDirichletState(Configuration conf) throws NoSuchMethodException,
+ InvocationTargetException {
String statePath = conf.get(DirichletDriver.STATE_IN_KEY);
String modelFactory = conf.get(DirichletDriver.MODEL_FACTORY_KEY);
String modelPrototype = conf.get(DirichletDriver.MODEL_PROTOTYPE_KEY);
String prototypeSize = conf.get(DirichletDriver.PROTOTYPE_SIZE_KEY);
String numClusters = conf.get(DirichletDriver.NUM_CLUSTERS_KEY);
String alpha0 = conf.get(DirichletDriver.ALPHA_0_KEY);
-
+
try {
- double alpha = Double.parseDouble(alpha0);
- DirichletState<VectorWritable> state = DirichletDriver.createState(modelFactory, modelPrototype,
- Integer.parseInt(prototypeSize), Integer.parseInt(numClusters), alpha);
- Path path = new Path(statePath);
- FileSystem fs = FileSystem.get(path.toUri(), conf);
- FileStatus[] status = fs.listStatus(path, new OutputLogFilter());
- for (FileStatus s : status) {
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
- try {
- Text key = new Text();
- DirichletCluster<VectorWritable> cluster = new DirichletCluster<VectorWritable>();
- while (reader.next(key, cluster)) {
- int index = Integer.parseInt(key.toString());
- state.getClusters().set(index, cluster);
- cluster = new DirichletCluster<VectorWritable>();
- }
- } finally {
- reader.close();
- }
- }
- // TODO: with more than one mapper, they will all have different mixtures. Will this matter?
- state.setMixture(UncommonDistributions.rDirichlet(state.totalCounts(), alpha));
- return state;
+ return loadState(conf,
+ statePath,
+ modelFactory,
+ modelPrototype,
+ Double.parseDouble(alpha0),
+ Integer.parseInt(prototypeSize),
+ Integer.parseInt(numClusters));
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e);
} catch (InstantiationException e) {
@@ -113,29 +97,51 @@ public class DirichletMapper extends Map
throw new IllegalStateException(e);
}
}
-
+
/**
- * Compute a normalized vector of probabilities that v is described by each model using the mixture and the
- * model pdfs
- *
- * @param state
- * the DirichletState<Vector> of this iteration
- * @param v
- * an Vector
- * @return the Vector of probabilities
+ * @param conf
+ * @param statePath
+ * @param modelFactory
+ * @param modelPrototype
+ * @param alpha
+ * @param pSize
+ * @param k
+ * @return
+ * @throws ClassNotFoundException
+ * @throws InstantiationException
+ * @throws IllegalAccessException
+ * @throws NoSuchMethodException
+ * @throws InvocationTargetException
+ * @throws IOException
*/
- private static Vector normalizedProbabilities(DirichletState<VectorWritable> state, VectorWritable v) {
- Vector pi = new DenseVector(state.getNumClusters());
- double max = 0;
- for (int k = 0; k < state.getNumClusters(); k++) {
- double p = state.adjustedProbability(v, k);
- pi.set(k, p);
- if (max < p) {
- max = p;
+ protected static DirichletState<VectorWritable> loadState(Configuration conf,
+ String statePath,
+ String modelFactory,
+ String modelPrototype,
+ double alpha,
+ int pSize,
+ int k) throws ClassNotFoundException, InstantiationException,
+ IllegalAccessException, NoSuchMethodException, InvocationTargetException, IOException {
+ DirichletState<VectorWritable> state = DirichletDriver.createState(modelFactory, modelPrototype, pSize, k, alpha);
+ Path path = new Path(statePath);
+ FileSystem fs = FileSystem.get(path.toUri(), conf);
+ FileStatus[] status = fs.listStatus(path, new OutputLogFilter());
+ for (FileStatus s : status) {
+ SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
+ try {
+ Text key = new Text();
+ DirichletCluster<VectorWritable> cluster = new DirichletCluster<VectorWritable>();
+ while (reader.next(key, cluster)) {
+ int index = Integer.parseInt(key.toString());
+ state.getClusters().set(index, cluster);
+ cluster = new DirichletCluster<VectorWritable>();
+ }
+ } finally {
+ reader.close();
}
}
- // normalize the probabilities by largest observed value
- pi.assign(new TimesFunction(), 1.0 / max);
- return pi;
+ // TODO: with more than one mapper, they will all have different mixtures. Will this matter?
+ state.setMixture(UncommonDistributions.rDirichlet(state.totalCounts(), alpha));
+ return state;
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletReducer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletReducer.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletReducer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/dirichlet/DirichletReducer.java Thu Jul 22 01:28:42 2010
@@ -26,7 +26,7 @@ import org.apache.mahout.math.VectorWrit
public class DirichletReducer extends Reducer<Text, VectorWritable, Text, DirichletCluster<VectorWritable>> {
- private DirichletState<VectorWritable> state;
+ private DirichletClusterer<VectorWritable> clusterer;
private Model<VectorWritable>[] newModels;
@@ -38,7 +38,8 @@ public class DirichletReducer extends Re
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
try {
- state = DirichletMapper.getDirichletState(context.getConfiguration());
+ clusterer = new DirichletClusterer<VectorWritable>(DirichletMapper.getDirichletState(context.getConfiguration()));
+ this.newModels = clusterer.samplePosteriorModels();
} catch (NumberFormatException e) {
throw new IllegalStateException(e);
} catch (SecurityException e) {
@@ -50,7 +51,6 @@ public class DirichletReducer extends Re
} catch (InvocationTargetException e) {
throw new IllegalStateException(e);
}
- this.newModels = state.getModelFactory().sampleFromPosterior(state.getModels());
}
@Override
@@ -58,25 +58,18 @@ public class DirichletReducer extends Re
int k = Integer.parseInt(key.toString());
Model<VectorWritable> model = newModels[k];
for (VectorWritable value : values) {
- model.observe(value);
+ // only observe real points, not the empty placeholders emitted by each mapper
+ if (value.get().size() > 0) {
+ model.observe(value);
+ }
}
- model.computeParameters();
- DirichletCluster<VectorWritable> cluster = state.getClusters().get(k);
- cluster.setModel(model);
- }
-
- @Override
- protected void cleanup(Context context) throws IOException, InterruptedException {
- for (int i = 0; i < state.getNumClusters(); i++) {
- DirichletCluster<VectorWritable> cluster = state.getClusters().get(i);
- context.write(new Text(String.valueOf(i)), cluster);
- }
- super.cleanup(context);
+ DirichletCluster<VectorWritable> cluster = clusterer.updateCluster(model, k);
+ context.write(new Text(String.valueOf(k)), cluster);
}
public void setup(DirichletState<VectorWritable> state) {
- this.state = state;
- this.newModels = state.getModelFactory().sampleFromPosterior(state.getModels());
+ clusterer = new DirichletClusterer<VectorWritable>(state);
+ this.newModels = clusterer.samplePosteriorModels();
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansDriver.java?rev=966474&r1=966473&r2=966474&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansDriver.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansDriver.java Thu Jul 22 01:28:42 2010
@@ -61,7 +61,7 @@ public class FuzzyKMeansDriver extends A
}
/**
- * Run the job using supplied arguments
+ * Run the job using supplied arguments on a new driver instance (convenience)
*
* @param input
* the directory pathname for input points
@@ -129,7 +129,7 @@ public class FuzzyKMeansDriver extends A
addOption(DefaultOptionCreator.convergenceOption().create());
addOption(DefaultOptionCreator.maxIterationsOption().create());
addOption(DefaultOptionCreator.overwriteOption().create());
- addOption(M_OPTION, M_OPTION,"coefficient normalization factor, must be greater than 1", true);
+ addOption(M_OPTION, M_OPTION, "coefficient normalization factor, must be greater than 1", true);
addOption(DefaultOptionCreator.numReducersOption().create());
//TODO: addOption(DefaultOptionCreator.numMappersOption().create()); but how to set in new Job?
addOption(DefaultOptionCreator.clusteringOption().create());
@@ -177,64 +177,6 @@ public class FuzzyKMeansDriver extends A
}
/**
- * Run the full clustering job
- * @param input
- * @param clustersIn
- * @param output
- * @param measureClass
- * @param convergenceDelta
- * @param maxIterations
- * @param numReduceTasks
- * @param m
- * @param runClustering
- * @param emitMostLikely
- * @param threshold
- * @throws IOException
- * @throws ClassNotFoundException
- * @throws InterruptedException
- */
- private void job(Path input,
- Path clustersIn,
- Path output,
- String measureClass,
- double convergenceDelta,
- int maxIterations,
- int numReduceTasks,
- float m,
- boolean runClustering,
- boolean emitMostLikely,
- double threshold) throws IOException, ClassNotFoundException, InterruptedException {
- boolean converged = false;
- int iteration = 1;
-
- // iterate until the clusters converge
- while (!converged && (iteration <= maxIterations)) {
- log.info("Iteration {}", iteration);
-
- // point the output to a new directory per iteration
- Path clustersOut = new Path(output, Cluster.CLUSTERS_DIR + iteration);
- converged = runIteration(input, clustersIn, clustersOut, measureClass, convergenceDelta, numReduceTasks, iteration, m);
-
- // now point the input to the old output directory
- clustersIn = clustersOut;
- iteration++;
- }
-
- // now actually cluster the points if requested
- if (runClustering) {
- log.info("Clustering ");
- runClustering(input,
- clustersIn,
- new Path(output, Cluster.CLUSTERED_POINTS_DIR),
- measureClass,
- convergenceDelta,
- m,
- emitMostLikely,
- threshold);
- }
- }
-
- /**
* Run the iteration using supplied arguments
*
* @param input
@@ -307,6 +249,111 @@ public class FuzzyKMeansDriver extends A
}
/**
+ * Iterate over the input vectors to produce clusters and, if requested, use the
+ * results of the final iteration to cluster the input vectors.
+ *
+ * @param input
+ * the directory pathname for input points
+ * @param clustersIn
+ * the directory pathname for initial & computed clusters
+ * @param output
+ * the directory pathname for output points
+ * @param measureClass
+ * the classname of the DistanceMeasure
+ * @param convergenceDelta
+ * the convergence delta value
+ * @param maxIterations
+ * the maximum number of iterations
+ * @param numReduceTasks
+ * the number of reduce tasks
+ * @param m
+ * the fuzzification factor, see
+ * http://en.wikipedia.org/wiki/Data_clustering#Fuzzy_c-means_clustering
+ * @param runClustering
+ * true if points are to be clustered after iterations complete
+ * @param emitMostLikely
+ * a boolean if true emit only most likely cluster for each point
+ * @param threshold
+ * a double threshold value emits all clusters having greater pdf (emitMostLikely = false)
+ * @throws IOException
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
+ */
+ public void job(Path input,
+ Path clustersIn,
+ Path output,
+ String measureClass,
+ double convergenceDelta,
+ int maxIterations,
+ int numReduceTasks,
+ float m,
+ boolean runClustering,
+ boolean emitMostLikely,
+ double threshold) throws IOException, ClassNotFoundException, InterruptedException {
+ Path clustersOut = buildClusters(input, clustersIn, output, measureClass, convergenceDelta, maxIterations, numReduceTasks, m);
+ if (runClustering) {
+ log.info("Clustering ");
+ clusterData(input,
+ clustersOut,
+ new Path(output, Cluster.CLUSTERED_POINTS_DIR),
+ measureClass,
+ convergenceDelta,
+ m,
+ emitMostLikely,
+ threshold);
+ }
+ }
+
+ /**
+ * Iterate over the input vectors to produce cluster directories for each iteration
+ *
+ * @param input
+ * the directory pathname for input points
+ * @param clustersIn
+ * the directory pathname for initial & computed clusters
+ * @param output
+ * the directory pathname for output points
+ * @param measureClass
+ * the classname of the DistanceMeasure
+ * @param convergenceDelta
+ * the convergence delta value
+ * @param maxIterations
+ * the maximum number of iterations
+ * @param numReduceTasks
+ * the number of reduce tasks
+ * @param m
+ * the fuzzification factor, see
+ * http://en.wikipedia.org/wiki/Data_clustering#Fuzzy_c-means_clustering
+ * @return the Path of the final clusters directory
+ * @throws IOException
+ */
+ public Path buildClusters(Path input,
+ Path clustersIn,
+ Path output,
+ String measureClass,
+ double convergenceDelta,
+ int maxIterations,
+ int numReduceTasks,
+ float m) throws IOException {
+ boolean converged = false;
+ int iteration = 1;
+
+ // iterate until the clusters converge
+ while (!converged && (iteration <= maxIterations)) {
+ log.info("Iteration {}", iteration);
+
+ // point the output to a new directory per iteration
+ Path clustersOut = new Path(output, Cluster.CLUSTERS_DIR + iteration);
+ converged = runIteration(input, clustersIn, clustersOut, measureClass, convergenceDelta, numReduceTasks, iteration, m);
+
+ // now point the input to the old output directory
+ clustersIn = clustersOut;
+ iteration++;
+ }
+ return clustersIn;
+ }
+
+ /**
* Run the job using supplied arguments
*
* @param input
@@ -325,14 +372,14 @@ public class FuzzyKMeansDriver extends A
* a double threshold value emits all clusters having greater pdf (emitMostLikely = false)
* @throws IOException
*/
- private void runClustering(Path input,
- Path clustersIn,
- Path output,
- String measureClass,
- double convergenceDelta,
- float m,
- boolean emitMostLikely,
- double threshold) throws IOException, ClassNotFoundException, InterruptedException {
+ public void clusterData(Path input,
+ Path clustersIn,
+ Path output,
+ String measureClass,
+ double convergenceDelta,
+ float m,
+ boolean emitMostLikely,
+ double threshold) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
conf.set(FuzzyKMeansConfigKeys.CLUSTER_PATH_KEY, clustersIn.toString());