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/06/11 01:28:12 UTC
svn commit: r953496 - in /mahout/trunk:
core/src/main/java/org/apache/mahout/clustering/meanshift/
core/src/test/java/org/apache/mahout/clustering/meanshift/
examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/
Author: jeastman
Date: Thu Jun 10 23:28:12 2010
New Revision: 953496
URL: http://svn.apache.org/viewvc?rev=953496&view=rev
Log:
MAHOUT-167: Converted MeanShift to 0.20.2. LDA is already good. A few minor TODOs to go back for
Removed:
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/OutputDriver.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/OutputMapper.java
Modified:
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterMapper.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterer.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyCreatorMapper.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyDriver.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyMapper.java
mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyReducer.java
mahout/trunk/core/src/test/java/org/apache/mahout/clustering/meanshift/TestMeanShift.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputDriver.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputMapper.java
mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/Job.java
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterMapper.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterMapper.java Thu Jun 10 23:28:12 2010
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -28,42 +29,41 @@ 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.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.OutputLogFilter;
-import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.clustering.WeightedVectorWritable;
import org.apache.mahout.math.VectorWritable;
-public class MeanShiftCanopyClusterMapper extends MapReduceBase implements
- Mapper<WritableComparable<?>, MeanShiftCanopy, IntWritable, WeightedVectorWritable> {
+public class MeanShiftCanopyClusterMapper extends Mapper<WritableComparable<?>, MeanShiftCanopy, IntWritable, WeightedVectorWritable> {
private List<MeanShiftCanopy> canopies;
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void map(WritableComparable<?> key,
- MeanShiftCanopy canopy,
- OutputCollector<IntWritable, WeightedVectorWritable> output,
- Reporter reporter) throws IOException {
+ protected void map(WritableComparable<?> key, MeanShiftCanopy canopy, Context context) throws IOException, InterruptedException {
// canopies use canopyIds assigned when input vectors are processed as vectorIds too
int vectorId = canopy.getId();
for (MeanShiftCanopy msc : canopies) {
for (int containedId : msc.getBoundPoints().toList()) {
if (vectorId == containedId) {
- output.collect(new IntWritable(msc.getId()),
+ context.write(new IntWritable(msc.getId()),
new WeightedVectorWritable(1, new VectorWritable(canopy.getCenter())));
}
}
}
}
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#setup(org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void configure(JobConf job) {
- super.configure(job);
+ protected void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
try {
- canopies = getCanopies(job);
+ canopies = getCanopies(context.getConfiguration());
} catch (SecurityException e) {
throw new IllegalStateException(e);
} catch (IllegalArgumentException e) {
@@ -71,15 +71,15 @@ public class MeanShiftCanopyClusterMappe
}
}
- public static List<MeanShiftCanopy> getCanopies(JobConf job) {
- String statePath = job.get(MeanShiftCanopyDriver.STATE_IN_KEY);
+ public static List<MeanShiftCanopy> getCanopies(Configuration configuration) {
+ String statePath = configuration.get(MeanShiftCanopyDriver.STATE_IN_KEY);
List<MeanShiftCanopy> canopies = new ArrayList<MeanShiftCanopy>();
try {
Path path = new Path(statePath);
- FileSystem fs = FileSystem.get(path.toUri(), job);
+ FileSystem fs = FileSystem.get(path.toUri(), configuration);
FileStatus[] status = fs.listStatus(path, 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(), configuration);
try {
Text key = new Text();
MeanShiftCanopy canopy = new MeanShiftCanopy();
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterer.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyClusterer.java Thu Jun 10 23:28:12 2010
@@ -21,9 +21,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.mahout.common.distance.DistanceMeasure;
import org.apache.mahout.common.distance.EuclideanDistanceMeasure;
@@ -44,8 +44,8 @@ public class MeanShiftCanopyClusterer {
// the distance measure
private DistanceMeasure measure;
- public MeanShiftCanopyClusterer(JobConf job) {
- configure(job);
+ public MeanShiftCanopyClusterer(Configuration configuration) {
+ configure(configuration);
}
public MeanShiftCanopyClusterer(DistanceMeasure aMeasure, double aT1, double aT2, double aDelta) {
@@ -63,14 +63,14 @@ public class MeanShiftCanopyClusterer {
/**
* Configure the Canopy and its distance measure
*
- * @param job
+ * @param configuration
* the JobConf for this job
*/
- public void configure(JobConf job) {
+ public void configure(Configuration configuration) {
try {
- measure = Class.forName(job.get(MeanShiftCanopyConfigKeys.DISTANCE_MEASURE_KEY)).asSubclass(DistanceMeasure.class)
+ measure = Class.forName(configuration.get(MeanShiftCanopyConfigKeys.DISTANCE_MEASURE_KEY)).asSubclass(DistanceMeasure.class)
.newInstance();
- measure.configure(job);
+ measure.configure(configuration);
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e);
} catch (IllegalAccessException e) {
@@ -79,9 +79,9 @@ public class MeanShiftCanopyClusterer {
throw new IllegalStateException(e);
}
// nextCanopyId = 0; // never read?
- t1 = Double.parseDouble(job.get(MeanShiftCanopyConfigKeys.T1_KEY));
- t2 = Double.parseDouble(job.get(MeanShiftCanopyConfigKeys.T2_KEY));
- convergenceDelta = Double.parseDouble(job.get(MeanShiftCanopyConfigKeys.CLUSTER_CONVERGENCE_KEY));
+ t1 = Double.parseDouble(configuration.get(MeanShiftCanopyConfigKeys.T1_KEY));
+ t2 = Double.parseDouble(configuration.get(MeanShiftCanopyConfigKeys.T2_KEY));
+ convergenceDelta = Double.parseDouble(configuration.get(MeanShiftCanopyConfigKeys.CLUSTER_CONVERGENCE_KEY));
}
/**
@@ -132,8 +132,7 @@ public class MeanShiftCanopyClusterer {
}
/** Emit the new canopy to the collector, keyed by the canopy's Id */
- static void emitCanopy(MeanShiftCanopy canopy,
- OutputCollector<Text, WritableComparable<?>> collector) throws IOException {
+ static void emitCanopy(MeanShiftCanopy canopy, OutputCollector<Text, WritableComparable<?>> collector) throws IOException {
String identifier = canopy.getIdentifier();
collector.collect(new Text(identifier), new Text("new " + canopy.toString()));
}
@@ -214,12 +213,8 @@ public class MeanShiftCanopyClusterer {
* @param numIter
* the maximum number of iterations
*/
- public static List<MeanShiftCanopy> clusterPoints(List<Vector> points,
- DistanceMeasure measure,
- double convergenceThreshold,
- double t1,
- double t2,
- int numIter) {
+ public static List<MeanShiftCanopy> clusterPoints(List<Vector> points, DistanceMeasure measure, double convergenceThreshold,
+ double t1, double t2, int numIter) {
MeanShiftCanopyClusterer clusterer = new MeanShiftCanopyClusterer(measure, t1, t2, convergenceThreshold);
List<MeanShiftCanopy> canopies = new ArrayList<MeanShiftCanopy>();
@@ -242,8 +237,7 @@ public class MeanShiftCanopyClusterer {
* @param canopies
* the List<MeanShiftCanopy> clusters
*/
- public static boolean runMeanShiftCanopyIteration(List<MeanShiftCanopy> canopies,
- MeanShiftCanopyClusterer clusterer) {
+ public static boolean runMeanShiftCanopyIteration(List<MeanShiftCanopy> canopies, MeanShiftCanopyClusterer clusterer) {
boolean converged = true;
List<MeanShiftCanopy> migratedCanopies = new ArrayList<MeanShiftCanopy>();
for (MeanShiftCanopy canopy : canopies) {
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyCreatorMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyCreatorMapper.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyCreatorMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyCreatorMapper.java Thu Jun 10 23:28:12 2010
@@ -22,36 +22,34 @@ import java.util.regex.Pattern;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.math.VectorWritable;
-public class MeanShiftCanopyCreatorMapper extends MapReduceBase implements
- Mapper<WritableComparable<?>,VectorWritable,Text,MeanShiftCanopy> {
+public class MeanShiftCanopyCreatorMapper extends Mapper<WritableComparable<?>, VectorWritable, Text, MeanShiftCanopy> {
+
+ private static final Pattern UNDERSCORE_PATTERN = Pattern.compile("_");
- private static final Pattern UNDERSCORE_PATTERN = Pattern.compile("_");
private static int nextCanopyId = -1;
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void map(WritableComparable<?> key,
- VectorWritable vector,
- OutputCollector<Text,MeanShiftCanopy> output,
- Reporter reporter) throws IOException {
- MeanShiftCanopy canopy = new MeanShiftCanopy(vector.get(), nextCanopyId++);
- output.collect(new Text(key.toString()), canopy);
+ protected void map(WritableComparable<?> key, VectorWritable point, Context context) throws IOException, InterruptedException {
+ MeanShiftCanopy canopy = new MeanShiftCanopy(point.get(), nextCanopyId++);
+ context.write(new Text(key.toString()), canopy);
}
-
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#setup(org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void configure(JobConf job) {
- super.configure(job);
+ protected void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
if (nextCanopyId == -1) {
- String taskId = job.get("mapred.task.id");
+ String taskId = context.getConfiguration().get("mapred.task.id");
String[] parts = UNDERSCORE_PATTERN.split(taskId);
- if (parts.length != 6 || !parts[0].equals("attempt")
- || (!"m".equals(parts[3]) && !"r".equals(parts[3]))) {
+ if (parts.length != 6 || !parts[0].equals("attempt") || (!"m".equals(parts[3]) && !"r".equals(parts[3]))) {
throw new IllegalArgumentException("TaskAttemptId string : " + taskId + " is not properly formed");
}
nextCanopyId = ((1 << 31) / 50000) * (Integer.parseInt(parts[4]));
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyDriver.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyDriver.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyDriver.java Thu Jun 10 23:28:12 2010
@@ -25,21 +25,18 @@ import org.apache.commons.cli2.Option;
import org.apache.commons.cli2.OptionException;
import org.apache.commons.cli2.builder.GroupBuilder;
import org.apache.commons.cli2.commandline.Parser;
-import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.FileInputFormat;
-import org.apache.hadoop.mapred.FileOutputFormat;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.SequenceFileInputFormat;
-import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+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.WeightedVectorWritable;
-import org.apache.mahout.clustering.fuzzykmeans.FuzzyKMeansDriver;
import org.apache.mahout.common.CommandLineUtil;
import org.apache.mahout.common.HadoopUtil;
import org.apache.mahout.common.commandline.DefaultOptionCreator;
@@ -57,7 +54,7 @@ public final class MeanShiftCanopyDriver
private MeanShiftCanopyDriver() {
}
- public static void main(String[] args) throws IOException {
+ public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Option inputOpt = DefaultOptionCreator.inputOption().create();
Option outputOpt = DefaultOptionCreator.outputOption().create();
Option convergenceDeltaOpt = DefaultOptionCreator.convergenceOption().create();
@@ -70,10 +67,9 @@ public final class MeanShiftCanopyDriver
Option threshold2Opt = DefaultOptionCreator.t2Option().create();
Option clusteringOpt = DefaultOptionCreator.clusteringOption().create();
- Group group = new GroupBuilder().withName("Options").withOption(inputOpt).withOption(outputOpt)
- .withOption(overwriteOutput).withOption(measureClassOpt).withOption(helpOpt)
- .withOption(convergenceDeltaOpt).withOption(threshold1Opt).withOption(threshold2Opt)
- .withOption(clusteringOpt).withOption(maxIterOpt).withOption(inputIsCanopiesOpt).create();
+ Group group = new GroupBuilder().withName("Options").withOption(inputOpt).withOption(outputOpt).withOption(overwriteOutput)
+ .withOption(measureClassOpt).withOption(helpOpt).withOption(convergenceDeltaOpt).withOption(threshold1Opt).withOption(
+ threshold2Opt).withOption(clusteringOpt).withOption(maxIterOpt).withOption(inputIsCanopiesOpt).create();
try {
Parser parser = new Parser();
@@ -95,15 +91,8 @@ public final class MeanShiftCanopyDriver
double t2 = Double.parseDouble(cmdLine.getValue(threshold2Opt).toString());
double convergenceDelta = Double.parseDouble(cmdLine.getValue(convergenceDeltaOpt).toString());
int maxIterations = Integer.parseInt(cmdLine.getValue(maxIterOpt).toString());
- runJob(input,
- output,
- measureClass,
- t1,
- t2,
- convergenceDelta,
- maxIterations,
- cmdLine.hasOption(inputIsCanopiesOpt),
- cmdLine.hasOption(clusteringOpt));
+ runJob(input, output, measureClass, t1, t2, convergenceDelta, maxIterations, cmdLine.hasOption(inputIsCanopiesOpt), cmdLine
+ .hasOption(clusteringOpt));
} catch (OptionException e) {
log.error("Exception parsing command line: ", e);
CommandLineUtil.printHelp(group);
@@ -127,60 +116,51 @@ public final class MeanShiftCanopyDriver
* the T2 distance threshold
* @param convergenceDelta
* the double convergence criteria
+ * @throws IOException
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
*/
static void runIteration(Path input, Path output, Path control, String measureClassName, double t1, double t2,
- double convergenceDelta) {
+ double convergenceDelta) throws IOException, InterruptedException, ClassNotFoundException {
- Configurable client = new JobClient();
- JobConf conf = new JobConf(MeanShiftCanopyDriver.class);
-
- conf.setOutputKeyClass(Text.class);
- conf.setOutputValueClass(MeanShiftCanopy.class);
-
- FileInputFormat.setInputPaths(conf, input);
- FileOutputFormat.setOutputPath(conf, output);
-
- conf.setMapperClass(MeanShiftCanopyMapper.class);
- conf.setReducerClass(MeanShiftCanopyReducer.class);
- conf.setNumReduceTasks(1);
- conf.setInputFormat(SequenceFileInputFormat.class);
- conf.setOutputFormat(SequenceFileOutputFormat.class);
+ Configuration conf = new Configuration();
conf.set(MeanShiftCanopyConfigKeys.DISTANCE_MEASURE_KEY, measureClassName);
conf.set(MeanShiftCanopyConfigKeys.CLUSTER_CONVERGENCE_KEY, String.valueOf(convergenceDelta));
conf.set(MeanShiftCanopyConfigKeys.T1_KEY, String.valueOf(t1));
conf.set(MeanShiftCanopyConfigKeys.T2_KEY, String.valueOf(t2));
conf.set(MeanShiftCanopyConfigKeys.CONTROL_PATH_KEY, control.toString());
- client.setConf(conf);
- try {
- JobClient.runJob(conf);
- } catch (IOException e) {
- log.warn(e.toString(), e);
- }
- }
+ Job job = new Job(conf);
- static void createCanopyFromVectors(Path input, Path output) {
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(MeanShiftCanopy.class);
- Configurable client = new JobClient();
- JobConf conf = new JobConf(MeanShiftCanopyDriver.class);
+ FileInputFormat.setInputPaths(job, input);
+ FileOutputFormat.setOutputPath(job, output);
- conf.setOutputKeyClass(Text.class);
- conf.setOutputValueClass(MeanShiftCanopy.class);
+ job.setMapperClass(MeanShiftCanopyMapper.class);
+ job.setReducerClass(MeanShiftCanopyReducer.class);
+ job.setNumReduceTasks(1);
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
- FileInputFormat.setInputPaths(conf, input);
- FileOutputFormat.setOutputPath(conf, output);
+ job.waitForCompletion(true);
+ }
- conf.setMapperClass(MeanShiftCanopyCreatorMapper.class);
- conf.setNumReduceTasks(0);
- conf.setInputFormat(SequenceFileInputFormat.class);
- conf.setOutputFormat(SequenceFileOutputFormat.class);
+ static void createCanopyFromVectors(Path input, Path output) throws IOException, InterruptedException, ClassNotFoundException {
+ Configuration conf = new Configuration();
+ Job job = new Job(conf);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(MeanShiftCanopy.class);
+ job.setMapperClass(MeanShiftCanopyCreatorMapper.class);
+ job.setNumReduceTasks(0);
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
- client.setConf(conf);
- try {
- JobClient.runJob(conf);
- } catch (IOException e) {
- log.warn(e.toString(), e);
- }
+ FileInputFormat.setInputPaths(job, input);
+ FileOutputFormat.setOutputPath(job, output);
+
+ job.waitForCompletion(true);
}
/**
@@ -192,32 +172,29 @@ public final class MeanShiftCanopyDriver
* the directory pathname for input clusters
* @param output
* the directory pathname for output clustered points
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
+ * @throws IOException
*/
- static void runClustering(Path input, Path clustersIn, Path output) {
-
- JobConf conf = new JobConf(FuzzyKMeansDriver.class);
- conf.setJobName("Mean Shift Clustering");
+ static void runClustering(Path input, Path clustersIn, Path output) throws IOException, InterruptedException,
+ ClassNotFoundException {
- conf.setOutputKeyClass(IntWritable.class);
- conf.setOutputValueClass(WeightedVectorWritable.class);
+ Configuration conf = new Configuration();
+ conf.set(STATE_IN_KEY, clustersIn.toString());
+ Job job = new Job(conf);
- FileInputFormat.setInputPaths(conf, input);
- FileOutputFormat.setOutputPath(conf, output);
+ job.setOutputKeyClass(IntWritable.class);
+ job.setOutputValueClass(WeightedVectorWritable.class);
+ job.setMapperClass(MeanShiftCanopyClusterMapper.class);
- conf.setMapperClass(MeanShiftCanopyClusterMapper.class);
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setNumReduceTasks(0);
- conf.setInputFormat(SequenceFileInputFormat.class);
- conf.setOutputFormat(SequenceFileOutputFormat.class);
+ FileInputFormat.setInputPaths(job, input);
+ FileOutputFormat.setOutputPath(job, output);
- // uncomment it to run locally
- // conf.set("mapred.job.tracker", "local");
- conf.setNumReduceTasks(0);
- conf.set(STATE_IN_KEY, clustersIn.toString());
- try {
- JobClient.runJob(conf);
- } catch (IOException e) {
- log.warn(e.toString(), e);
- }
+ job.waitForCompletion(true);
}
/**
@@ -241,19 +218,12 @@ public final class MeanShiftCanopyDriver
true if the input path already contains MeanShiftCanopies and does not need to be converted from Vectors
* @param runClustering
* true if the input points are to be clustered once the iterations complete
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
*/
- public static void runJob(Path input,
- Path output,
- String measureClassName,
- double t1,
- double t2,
- double convergenceDelta,
- int maxIterations,
- boolean inputIsCanopies,
- boolean runClustering) throws IOException {
- // delete the output directory
- Configuration conf = new JobConf(MeanShiftCanopyDriver.class);
-
+ public static void runJob(Path input, Path output, String measureClassName, double t1, double t2, double convergenceDelta,
+ int maxIterations, boolean inputIsCanopies, boolean runClustering) throws IOException, InterruptedException,
+ ClassNotFoundException {
Path clustersIn = new Path(output, Cluster.INITIAL_CLUSTERS_DIR);
if (inputIsCanopies) {
clustersIn = input;
@@ -270,7 +240,7 @@ public final class MeanShiftCanopyDriver
Path clustersOut = new Path(output, Cluster.CLUSTERS_DIR + iteration);
Path controlOut = new Path(output, CONTROL_CONVERGED);
runIteration(clustersIn, clustersOut, controlOut, measureClassName, t1, t2, convergenceDelta);
- converged = FileSystem.get(conf).exists(controlOut);
+ converged = FileSystem.get(new Configuration()).exists(controlOut);
// now point the input to the old output directory
clustersIn = clustersOut;
iteration++;
@@ -278,10 +248,8 @@ public final class MeanShiftCanopyDriver
if (runClustering) {
// now cluster the points
- runClustering(
- inputIsCanopies ? input : new Path(output, Cluster.INITIAL_CLUSTERS_DIR),
- clustersIn,
- new Path(output, Cluster.CLUSTERED_POINTS_DIR));
+ runClustering(inputIsCanopies ? input : new Path(output, Cluster.INITIAL_CLUSTERS_DIR), clustersIn, new Path(output,
+ Cluster.CLUSTERED_POINTS_DIR));
}
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyMapper.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyMapper.java Thu Jun 10 23:28:12 2010
@@ -23,42 +23,40 @@ import java.util.List;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.Mapper;
-public class MeanShiftCanopyMapper extends MapReduceBase implements
- Mapper<WritableComparable<?>,MeanShiftCanopy,Text,MeanShiftCanopy> {
+public class MeanShiftCanopyMapper extends Mapper<WritableComparable<?>,MeanShiftCanopy,Text,MeanShiftCanopy> {
private final List<MeanShiftCanopy> canopies = new ArrayList<MeanShiftCanopy>();
private MeanShiftCanopyClusterer clusterer;
- private OutputCollector<Text,MeanShiftCanopy> output;
-
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#setup(org.apache.hadoop.mapreduce.Mapper.Context)
+ */
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+ clusterer = new MeanShiftCanopyClusterer(context.getConfiguration());
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void map(WritableComparable<?> key,
- MeanShiftCanopy canopy,
- OutputCollector<Text,MeanShiftCanopy> output,
- Reporter reporter) throws IOException {
- this.output = output;
+ protected void map(WritableComparable<?> key, MeanShiftCanopy canopy, Context context) throws IOException, InterruptedException {
clusterer.mergeCanopy(canopy.shallowCopy(), canopies);
}
-
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#cleanup(org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void close() throws IOException {
+ protected void cleanup(Context context) throws IOException, InterruptedException {
for (MeanShiftCanopy canopy : canopies) {
clusterer.shiftToMean(canopy);
- output.collect(new Text("canopy"), canopy);
+ context.write(new Text("canopy"), canopy);
}
- super.close();
+ super.cleanup(context);
}
-
- @Override
- public void configure(JobConf job) {
- super.configure(job);
- clusterer = new MeanShiftCanopyClusterer(job);
- }
-
+
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyReducer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyReducer.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyReducer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/meanshift/MeanShiftCanopyReducer.java Thu Jun 10 23:28:12 2010
@@ -22,60 +22,59 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reducer;
-import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.Reducer;
-public class MeanShiftCanopyReducer extends MapReduceBase implements
- Reducer<Text,MeanShiftCanopy,Text,MeanShiftCanopy> {
+public class MeanShiftCanopyReducer extends Reducer<Text,MeanShiftCanopy,Text,MeanShiftCanopy> {
private final List<MeanShiftCanopy> canopies = new ArrayList<MeanShiftCanopy>();
private MeanShiftCanopyClusterer clusterer;
private boolean allConverged = true;
- private JobConf conf;
-
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Reducer#setup(org.apache.hadoop.mapreduce.Reducer.Context)
+ */
@Override
- public void reduce(Text key,
- Iterator<MeanShiftCanopy> values,
- OutputCollector<Text,MeanShiftCanopy> output,
- Reporter reporter) throws IOException {
-
- while (values.hasNext()) {
- MeanShiftCanopy canopy = values.next();
+ protected void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+ clusterer = new MeanShiftCanopyClusterer(context.getConfiguration());
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Reducer#reduce(java.lang.Object, java.lang.Iterable, org.apache.hadoop.mapreduce.Reducer.Context)
+ */
+ @Override
+ protected void reduce(Text key, Iterable<MeanShiftCanopy> values, Context context) throws IOException, InterruptedException {
+ Iterator<MeanShiftCanopy> it = values.iterator();
+ while (it.hasNext()) {
+ MeanShiftCanopy canopy = it.next();
clusterer.mergeCanopy(canopy.shallowCopy(), canopies);
}
for (MeanShiftCanopy canopy : canopies) {
boolean converged = clusterer.shiftToMean(canopy);
if (converged) {
- reporter.incrCounter("Clustering", "Converged Clusters", 1);
+ // TODO: reporter.incrCounter("Clustering", "Converged Clusters", 1);
}
allConverged = converged && allConverged;
- output.collect(new Text(canopy.getIdentifier()), canopy);
+ context.write(new Text(canopy.getIdentifier()), canopy);
}
}
-
- @Override
- public void configure(JobConf job) {
- super.configure(job);
- this.conf = job;
- clusterer = new MeanShiftCanopyClusterer(job);
- }
-
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Reducer#cleanup(org.apache.hadoop.mapreduce.Reducer.Context)
+ */
@Override
- public void close() throws IOException {
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ Configuration conf = context.getConfiguration();
if (allConverged) {
Path path = new Path(conf.get(MeanShiftCanopyConfigKeys.CONTROL_PATH_KEY));
FileSystem.get(conf).createNewFile(path);
}
- super.close();
+ super.cleanup(context);
}
-
}
Modified: mahout/trunk/core/src/test/java/org/apache/mahout/clustering/meanshift/TestMeanShift.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/test/java/org/apache/mahout/clustering/meanshift/TestMeanShift.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/core/src/test/java/org/apache/mahout/clustering/meanshift/TestMeanShift.java (original)
+++ mahout/trunk/core/src/test/java/org/apache/mahout/clustering/meanshift/TestMeanShift.java Thu Jun 10 23:28:12 2010
@@ -27,10 +27,10 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.JobConf;
import org.apache.mahout.clustering.ClusteringTestUtils;
+import org.apache.mahout.clustering.MockMapperContext;
+import org.apache.mahout.clustering.MockReducerContext;
import org.apache.mahout.common.DummyOutputCollector;
-import org.apache.mahout.common.DummyReporter;
import org.apache.mahout.common.MahoutTestCase;
import org.apache.mahout.common.distance.DistanceMeasure;
import org.apache.mahout.common.distance.EuclideanDistanceMeasure;
@@ -143,8 +143,6 @@ public class TestMeanShift extends Mahou
* CanopyMapper/Combiner which clusters input points to produce an output set of canopies.
*/
public void testCanopyMapperEuclidean() throws Exception {
- MeanShiftCanopyMapper mapper = new MeanShiftCanopyMapper();
- DummyOutputCollector<Text, MeanShiftCanopy> collector = new DummyOutputCollector<Text, MeanShiftCanopy>();
MeanShiftCanopyClusterer clusterer = new MeanShiftCanopyClusterer(euclideanDistanceMeasure, 4, 1, 0.5);
// get the initial canopies
List<MeanShiftCanopy> canopies = getInitialCanopies();
@@ -155,22 +153,25 @@ public class TestMeanShift extends Mahou
clusterer.mergeCanopy(new MeanShiftCanopy(aRaw, nextCanopyId++), refCanopies);
}
- JobConf conf = new JobConf();
+ Configuration conf = new Configuration();
conf.set(MeanShiftCanopyConfigKeys.DISTANCE_MEASURE_KEY, "org.apache.mahout.common.distance.EuclideanDistanceMeasure");
conf.set(MeanShiftCanopyConfigKeys.T1_KEY, "4");
conf.set(MeanShiftCanopyConfigKeys.T2_KEY, "1");
conf.set(MeanShiftCanopyConfigKeys.CLUSTER_CONVERGENCE_KEY, "0.5");
- mapper.configure(conf);
// map the data
+ MeanShiftCanopyMapper mapper = new MeanShiftCanopyMapper();
+ DummyOutputCollector<Text, MeanShiftCanopy> mapCollector = new DummyOutputCollector<Text, MeanShiftCanopy>();
+ MockMapperContext<Text, MeanShiftCanopy> mapContext = new MockMapperContext<Text, MeanShiftCanopy>(mapper, conf, mapCollector);
+ mapper.setup(mapContext);
for (MeanShiftCanopy canopy : canopies) {
- mapper.map(new Text(), canopy, collector, null);
+ mapper.map(new Text(), canopy, mapContext);
}
- mapper.close();
+ mapper.cleanup(mapContext);
// now verify the output
- assertEquals("Number of map results", 1, collector.getData().size());
- List<MeanShiftCanopy> data = collector.getValue(new Text("canopy"));
+ assertEquals("Number of map results", 1, mapCollector.getData().size());
+ List<MeanShiftCanopy> data = mapCollector.getValue(new Text("canopy"));
assertEquals("Number of canopies", refCanopies.size(), data.size());
// add all points to the reference canopies
@@ -200,9 +201,6 @@ public class TestMeanShift extends Mahou
* clusters input centroid points to produce an output set of final canopy centroid points.
*/
public void testCanopyReducerEuclidean() throws Exception {
- MeanShiftCanopyMapper mapper = new MeanShiftCanopyMapper();
- MeanShiftCanopyReducer reducer = new MeanShiftCanopyReducer();
- DummyOutputCollector<Text, MeanShiftCanopy> mapCollector = new DummyOutputCollector<Text, MeanShiftCanopy>();
MeanShiftCanopyClusterer clusterer = new MeanShiftCanopyClusterer(euclideanDistanceMeasure, 4, 1, 0.5);
// get the initial canopies
List<MeanShiftCanopy> canopies = getInitialCanopies();
@@ -224,25 +222,32 @@ public class TestMeanShift extends Mahou
clusterer.shiftToMean(canopy);
}
- JobConf conf = new JobConf();
+ Configuration conf = new Configuration();
conf.set(MeanShiftCanopyConfigKeys.DISTANCE_MEASURE_KEY, "org.apache.mahout.common.distance.EuclideanDistanceMeasure");
conf.set(MeanShiftCanopyConfigKeys.T1_KEY, "4");
conf.set(MeanShiftCanopyConfigKeys.T2_KEY, "1");
conf.set(MeanShiftCanopyConfigKeys.CLUSTER_CONVERGENCE_KEY, "0.5");
- mapper.configure(conf);
+
+ MeanShiftCanopyMapper mapper = new MeanShiftCanopyMapper();
+ DummyOutputCollector<Text, MeanShiftCanopy> mapCollector = new DummyOutputCollector<Text, MeanShiftCanopy>();
+ MockMapperContext<Text, MeanShiftCanopy> mapContext = new MockMapperContext<Text, MeanShiftCanopy>(mapper, conf, mapCollector);
+ mapper.setup(mapContext);
// map the data
for (MeanShiftCanopy canopy : canopies) {
- mapper.map(new Text(), canopy, mapCollector, null);
+ mapper.map(new Text(), canopy, mapContext);
}
- mapper.close();
+ mapper.cleanup(mapContext);
assertEquals("Number of map results", 1, mapCollector.getData().size());
// now reduce the mapper output
+ MeanShiftCanopyReducer reducer = new MeanShiftCanopyReducer();
DummyOutputCollector<Text, MeanShiftCanopy> reduceCollector = new DummyOutputCollector<Text, MeanShiftCanopy>();
- reducer.configure(conf);
- reducer.reduce(new Text("canopy"), mapCollector.getValue(new Text("canopy")).iterator(), reduceCollector, new DummyReporter());
- reducer.close();
+ MockReducerContext<Text, MeanShiftCanopy> reduceContext = new MockReducerContext<Text, MeanShiftCanopy>(reducer, conf,
+ reduceCollector, Text.class, MeanShiftCanopy.class);
+ reducer.setup(reduceContext);
+ reducer.reduce(new Text("canopy"), mapCollector.getValue(new Text("canopy")), reduceContext);
+ reducer.cleanup(reduceContext);
// now verify the output
assertEquals("Number of canopies", reducerReference.size(), reduceCollector.getKeys().size());
@@ -287,8 +292,8 @@ public class TestMeanShift extends Mahou
// now run the Job
Path output = getTestTempDirPath("output");
MeanShiftCanopyDriver.runJob(input, output, EuclideanDistanceMeasure.class.getName(), 4, 1, 0.5, 10, false, false);
- JobConf conf = new JobConf(MeanShiftCanopyDriver.class);
- Path outPart = new Path(output, "clusters-3/part-00000");
+ Configuration conf = new Configuration();
+ Path outPart = new Path(output, "clusters-3/part-r-00000");
SequenceFile.Reader reader = new SequenceFile.Reader(fs, outPart, conf);
Text key = new Text();
MeanShiftCanopy value = new MeanShiftCanopy();
Modified: mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputDriver.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputDriver.java (original)
+++ mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputDriver.java Thu Jun 10 23:28:12 2010
@@ -25,14 +25,14 @@ import org.apache.commons.cli2.Option;
import org.apache.commons.cli2.OptionException;
import org.apache.commons.cli2.builder.GroupBuilder;
import org.apache.commons.cli2.commandline.Parser;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.FileInputFormat;
-import org.apache.hadoop.mapred.FileOutputFormat;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.Reducer;
-import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.mahout.clustering.meanshift.MeanShiftCanopy;
import org.apache.mahout.common.CommandLineUtil;
import org.apache.mahout.common.commandline.DefaultOptionCreator;
@@ -42,18 +42,18 @@ import org.slf4j.LoggerFactory;
public final class InputDriver {
private static final Logger LOG = LoggerFactory.getLogger(InputDriver.class);
-
- private InputDriver() { }
-
- public static void main(String[] args) throws IOException {
+
+ private InputDriver() {
+ }
+
+ public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
GroupBuilder gbuilder = new GroupBuilder();
-
+
Option inputOpt = DefaultOptionCreator.inputOption().withRequired(false).create();
Option outputOpt = DefaultOptionCreator.outputOption().withRequired(false).create();
Option helpOpt = DefaultOptionCreator.helpOption();
- Group group = gbuilder.withName("Options").withOption(inputOpt).withOption(outputOpt).withOption(helpOpt)
- .create();
-
+ Group group = gbuilder.withName("Options").withOption(inputOpt).withOption(outputOpt).withOption(helpOpt).create();
+
try {
Parser parser = new Parser();
parser.setGroup(group);
@@ -62,7 +62,7 @@ public final class InputDriver {
CommandLineUtil.printHelp(group);
return;
}
-
+
Path input = new Path(cmdLine.getValue(inputOpt, "testdata").toString());
Path output = new Path(cmdLine.getValue(outputOpt, "output").toString());
runJob(input, output);
@@ -71,23 +71,21 @@ public final class InputDriver {
CommandLineUtil.printHelp(group);
}
}
-
- public static void runJob(Path input, Path output) throws IOException {
- JobClient client = new JobClient();
- JobConf conf = new JobConf(org.apache.mahout.clustering.syntheticcontrol.meanshift.InputDriver.class);
-
- conf.setOutputKeyClass(Text.class);
- conf.setOutputValueClass(MeanShiftCanopy.class);
-
- FileInputFormat.setInputPaths(conf, input);
- FileOutputFormat.setOutputPath(conf, output);
- conf.setOutputFormat(SequenceFileOutputFormat.class);
- conf.setMapperClass(org.apache.mahout.clustering.syntheticcontrol.meanshift.InputMapper.class);
- conf.setReducerClass(Reducer.class);
- conf.setNumReduceTasks(0);
-
- client.setConf(conf);
- JobClient.runJob(conf);
+
+ public static void runJob(Path input, Path output) throws IOException, InterruptedException, ClassNotFoundException {
+ Configuration conf = new Configuration();
+
+ Job job = new Job(conf);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(MeanShiftCanopy.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setMapperClass(org.apache.mahout.clustering.syntheticcontrol.meanshift.InputMapper.class);
+ job.setReducerClass(Reducer.class);
+ job.setNumReduceTasks(0);
+
+ FileInputFormat.setInputPaths(job, input);
+ FileOutputFormat.setOutputPath(job, output);
+
+ job.waitForCompletion(true);
}
-
}
Modified: mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputMapper.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputMapper.java (original)
+++ mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/InputMapper.java Thu Jun 10 23:28:12 2010
@@ -24,24 +24,22 @@ import java.util.regex.Pattern;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.clustering.meanshift.MeanShiftCanopy;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.Vector;
-public class InputMapper extends MapReduceBase implements Mapper<LongWritable,Text,Text,MeanShiftCanopy> {
+public class InputMapper extends Mapper<LongWritable,Text,Text,MeanShiftCanopy> {
private static final Pattern SPACE = Pattern.compile(" ");
private int nextCanopyId;
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, org.apache.hadoop.mapreduce.Mapper.Context)
+ */
@Override
- public void map(LongWritable key,
- Text values,
- OutputCollector<Text,MeanShiftCanopy> output,
- Reporter reporter) throws IOException {
+ protected void map(LongWritable key, Text values, Context context) throws IOException, InterruptedException {
String[] numbers = InputMapper.SPACE.split(values.toString());
// sometimes there are multiple separator spaces
List<Double> doubles = new ArrayList<Double>();
@@ -56,7 +54,6 @@ public class InputMapper extends MapRedu
point.set(index++, d);
}
MeanShiftCanopy canopy = new MeanShiftCanopy(point, nextCanopyId++);
- output.collect(new Text(), canopy);
+ context.write(new Text(), canopy);
}
-
}
Modified: mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/Job.java
URL: http://svn.apache.org/viewvc/mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/Job.java?rev=953496&r1=953495&r2=953496&view=diff
==============================================================================
--- mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/Job.java (original)
+++ mahout/trunk/examples/src/main/java/org/apache/mahout/clustering/syntheticcontrol/meanshift/Job.java Thu Jun 10 23:28:12 2010
@@ -27,9 +27,9 @@ import org.apache.commons.cli2.builder.A
import org.apache.commons.cli2.builder.DefaultOptionBuilder;
import org.apache.commons.cli2.builder.GroupBuilder;
import org.apache.commons.cli2.commandline.Parser;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
import org.apache.mahout.clustering.meanshift.MeanShiftCanopyDriver;
import org.apache.mahout.clustering.syntheticcontrol.Constants;
import org.apache.mahout.common.CommandLineUtil;
@@ -119,6 +119,8 @@ public final class Job {
* the double convergence criteria for iterations
* @param maxIterations
* the int maximum number of iterations
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
* @throws IllegalAccessException
* @throws InstantiationException
*/
@@ -128,9 +130,9 @@ public final class Job {
double t1,
double t2,
double convergenceDelta,
- int maxIterations) throws IOException {
+ int maxIterations) throws IOException, InterruptedException, ClassNotFoundException {
JobClient client = new JobClient();
- JobConf conf = new JobConf(Job.class);
+ Configuration conf = new Configuration();
client.setConf(conf);
HadoopUtil.overwriteOutput(output);