You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by sr...@apache.org on 2011/03/31 11:25:30 UTC
svn commit: r1087225 [2/7] - in /mahout/trunk:
core/src/main/java/org/apache/mahout/cf/taste/hadoop/
core/src/main/java/org/apache/mahout/cf/taste/hadoop/item/
core/src/main/java/org/apache/mahout/cf/taste/hadoop/pseudo/
core/src/main/java/org/apache/m...
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesThetaMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesThetaMapper.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesThetaMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesThetaMapper.java Thu Mar 31 09:25:25 2011
@@ -22,14 +22,14 @@ import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
-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.Writable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.classifier.naivebayes.BayesConstants;
+import org.apache.mahout.common.Pair;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileIterable;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.apache.mahout.math.map.OpenObjectIntHashMap;
@@ -55,41 +55,32 @@ public class NaiveBayesThetaMapper exten
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
Configuration conf = context.getConfiguration();
- try {
- URI[] localFiles = DistributedCache.getCacheFiles(conf);
- if (localFiles == null || localFiles.length < 2) {
- throw new IllegalArgumentException("missing paths from the DistributedCache");
- }
- alphaI = conf.getFloat(NaiveBayesTrainer.ALPHA_I, 1.0f);
- Path weightFile = new Path(localFiles[0].getPath());
- FileSystem fs = weightFile.getFileSystem(conf);
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, weightFile, conf);
- Writable key = new Text();
- VectorWritable value = new VectorWritable();
+ URI[] localFiles = DistributedCache.getCacheFiles(conf);
+ if (localFiles == null || localFiles.length < 2) {
+ throw new IllegalArgumentException("missing paths from the DistributedCache");
+ }
+ alphaI = conf.getFloat(NaiveBayesTrainer.ALPHA_I, 1.0f);
+ Path weightFile = new Path(localFiles[0].getPath());
- while (reader.next(key, value)) {
- if (key.toString().equals(BayesConstants.FEATURE_SUM)) {
- featureSum = value.get();
- } else if (key.toString().equals(BayesConstants.LABEL_SUM)) {
- labelSum = value.get();
- }
+ for (Pair<Text,VectorWritable> record :
+ new SequenceFileIterable<Text,VectorWritable>(weightFile, true, conf)) {
+ Text key = record.getFirst();
+ VectorWritable value = record.getSecond();
+ if (key.toString().equals(BayesConstants.FEATURE_SUM)) {
+ featureSum = value.get();
+ } else if (key.toString().equals(BayesConstants.LABEL_SUM)) {
+ labelSum = value.get();
}
- perLabelThetaNormalizer = labelSum.like();
- vocabCount = featureSum.getNumNondefaultElements();
-
- Path labelMapFile = new Path(localFiles[1].getPath());
- fs = labelMapFile.getFileSystem(conf);
+ }
+ perLabelThetaNormalizer = labelSum.like();
+ vocabCount = featureSum.getNumNondefaultElements();
- reader.close();
- reader = new SequenceFile.Reader(fs, labelMapFile, conf);
- IntWritable intValue = new IntWritable();
+ Path labelMapFile = new Path(localFiles[1].getPath());
- // key is word value is id
- while (reader.next(key, intValue)) {
- labelMap.put(key.toString(), intValue.get());
- }
- } catch (IOException e) {
- throw new IllegalStateException(e);
+ // key is word value is id
+ for (Pair<Writable,IntWritable> record :
+ new SequenceFileIterable<Writable,IntWritable>(labelMapFile, true, conf)) {
+ labelMap.put(record.getFirst().toString(), record.getSecond().get());
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesTrainer.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesTrainer.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesTrainer.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesTrainer.java Thu Mar 31 09:25:25 2011
@@ -99,7 +99,7 @@ public final class NaiveBayesTrainer {
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(VectorWritable.class);
job.setNumReduceTasks(numReducers);
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(conf, output);
job.waitForCompletion(true);
}
@@ -127,7 +127,7 @@ public final class NaiveBayesTrainer {
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(VectorWritable.class);
job.setNumReduceTasks(numReducers);
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(conf, output);
job.waitForCompletion(true);
}
@@ -156,7 +156,7 @@ public final class NaiveBayesTrainer {
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(VectorWritable.class);
job.setNumReduceTasks(numReducers);
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(conf, output);
job.waitForCompletion(true);
}
@@ -185,7 +185,7 @@ public final class NaiveBayesTrainer {
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(VectorWritable.class);
job.setNumReduceTasks(numReducers);
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(conf, output);
job.waitForCompletion(true);
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesWeightsMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesWeightsMapper.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesWeightsMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/classifier/naivebayes/trainer/NaiveBayesWeightsMapper.java Thu Mar 31 09:25:25 2011
@@ -22,14 +22,14 @@ import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
-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.Writable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.classifier.naivebayes.BayesConstants;
+import org.apache.mahout.common.Pair;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileIterable;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
@@ -58,23 +58,16 @@ public class NaiveBayesWeightsMapper ext
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
Configuration conf = context.getConfiguration();
- try {
- URI[] localFiles = DistributedCache.getCacheFiles(conf);
- if (localFiles == null || localFiles.length < 1) {
- throw new IllegalArgumentException("missing paths from the DistributedCache");
- }
- Path labelMapFile = new Path(localFiles[0].getPath());
- FileSystem fs = labelMapFile.getFileSystem(conf);
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, labelMapFile, conf);
- Writable key = new Text();
- IntWritable value = new IntWritable();
+ URI[] localFiles = DistributedCache.getCacheFiles(conf);
+ if (localFiles == null || localFiles.length < 1) {
+ throw new IllegalArgumentException("missing paths from the DistributedCache");
+ }
+ Path labelMapFile = new Path(localFiles[0].getPath());
- // key is word value is id
- while (reader.next(key, value)) {
- labelMap.put(key.toString(), value.get());
- }
- } catch (IOException e) {
- throw new IllegalStateException(e);
+ // key is word value is id
+ for (Pair<Writable,IntWritable> record :
+ new SequenceFileIterable<Writable,IntWritable>(labelMapFile, true, conf)) {
+ labelMap.put(record.getFirst().toString(), record.getSecond().get());
}
}
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=1087225&r1=1087224&r2=1087225&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 Mar 31 09:25:25 2011
@@ -29,7 +29,6 @@ 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.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -39,11 +38,14 @@ import org.apache.hadoop.util.ToolRunner
import org.apache.mahout.clustering.AbstractCluster;
import org.apache.mahout.clustering.Cluster;
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.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
import org.apache.mahout.math.VectorWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -78,7 +80,7 @@ public class CanopyDriver extends Abstra
Path input = getInputPath();
Path output = getOutputPath();
if (hasOption(DefaultOptionCreator.OVERWRITE_OPTION)) {
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(getConf(), output);
}
String measureClass = getOption(DefaultOptionCreator.DISTANCE_MEASURE_OPTION);
double t1 = Double.parseDouble(getOption(DefaultOptionCreator.T1_OPTION));
@@ -110,8 +112,7 @@ public class CanopyDriver extends Abstra
double t1,
double t2,
boolean runClustering,
- boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException,
- InstantiationException, IllegalAccessException {
+ boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException {
Path clustersOut = buildClusters(conf, input, output, measure, t1, t2, runSequential);
if (runClustering) {
clusterData(conf, input, clustersOut, output, measure, t1, t2, runSequential);
@@ -135,8 +136,7 @@ public class CanopyDriver extends Abstra
double t1,
double t2,
boolean runClustering,
- boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException,
- InstantiationException, IllegalAccessException {
+ boolean runSequential) throws IOException, InterruptedException, ClassNotFoundException {
run(new Configuration(), input, output, measure, t1, t2, runClustering, runSequential);
}
@@ -160,7 +160,7 @@ public class CanopyDriver extends Abstra
double t1,
double t2,
boolean runSequential)
- throws InstantiationException, IllegalAccessException, IOException, InterruptedException, ClassNotFoundException {
+ throws IOException, InterruptedException, ClassNotFoundException {
log.info("Build Clusters Input: {} Out: {} " + "Measure: {} t1: {} t2: {}",
new Object[] { input, output, measure, t1, t2 });
if (runSequential) {
@@ -182,24 +182,14 @@ public class CanopyDriver extends Abstra
* @return the canopy output directory Path
*/
private static Path buildClustersSeq(Path input, Path output, DistanceMeasure measure, double t1, double t2)
- throws InstantiationException, IllegalAccessException, IOException {
+ throws IOException {
CanopyClusterer clusterer = new CanopyClusterer(measure, t1, t2);
Collection<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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.addPointToCanopies(vw.get(), canopies);
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- }
- } finally {
- reader.close();
- }
+ for (VectorWritable value :
+ new SequenceFileDirValueIterable<VectorWritable>(input, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ clusterer.addPointToCanopies(value.get(), canopies);
}
Path canopyOutputDir = new Path(output, Cluster.CLUSTERS_DIR + '0');
Path path = new Path(canopyOutputDir, "part-r-00000");
@@ -207,8 +197,13 @@ public class CanopyDriver extends Abstra
try {
for (Canopy canopy : canopies) {
canopy.computeParameters();
- log.debug("Writing Canopy:" + canopy.getIdentifier() + " center:" + AbstractCluster.formatVector(canopy.getCenter(), null)
- + " numPoints:" + canopy.getNumPoints() + " radius:" + AbstractCluster.formatVector(canopy.getRadius(), null));
+ log.debug("Writing Canopy:{} center:{} numPoints:{} radius:{}",
+ new Object[] {
+ canopy.getIdentifier(),
+ AbstractCluster.formatVector(canopy.getCenter(), null),
+ canopy.getNumPoints(),
+ AbstractCluster.formatVector(canopy.getRadius(), null)
+ });
writer.append(new Text(canopy.getIdentifier()), canopy);
}
} finally {
@@ -229,8 +224,12 @@ public class CanopyDriver extends Abstra
* @param t2 the double T2 distance metric
* @return the canopy output directory Path
*/
- private static Path buildClustersMR(Configuration conf, Path input, Path output, DistanceMeasure measure, double t1, double t2)
- throws IOException, InterruptedException, ClassNotFoundException {
+ private static Path buildClustersMR(Configuration conf,
+ Path input,
+ Path output,
+ DistanceMeasure measure,
+ double t1, double t2)
+ throws IOException, InterruptedException, ClassNotFoundException {
conf.set(CanopyConfigKeys.DISTANCE_MEASURE_KEY, measure.getClass().getName());
conf.set(CanopyConfigKeys.T1_KEY, String.valueOf(t1));
conf.set(CanopyConfigKeys.T2_KEY, String.valueOf(t2));
@@ -263,8 +262,8 @@ public class CanopyDriver extends Abstra
DistanceMeasure measure,
double t1,
double t2,
- boolean runSequential) throws InstantiationException, IllegalAccessException, IOException,
- InterruptedException, ClassNotFoundException {
+ boolean runSequential)
+ throws IOException, InterruptedException, ClassNotFoundException {
if (runSequential) {
clusterDataSeq(points, canopies, output, measure, t1, t2);
} else {
@@ -277,49 +276,32 @@ public class CanopyDriver extends Abstra
Path output,
DistanceMeasure measure,
double t1,
- double t2)
- throws InstantiationException, IllegalAccessException, IOException {
+ double t2) throws IOException {
CanopyClusterer clusterer = new CanopyClusterer(measure, t1, t2);
Collection<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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- Canopy value = reader.getValueClass().asSubclass(Canopy.class).newInstance();
- while (reader.next(key, value)) {
- clusters.add(value);
- value = reader.getValueClass().asSubclass(Canopy.class).newInstance();
- }
- } finally {
- reader.close();
- }
+ for (Canopy value :
+ new SequenceFileDirValueIterable<Canopy>(canopies, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ clusters.add(value);
}
// 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());
+ FileSystem fs = FileSystem.get(points.toUri(), conf);
+ FileStatus[] status = fs.listStatus(points, PathFilters.logsCRCFilter());
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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- Canopy closest = clusterer.findClosestCanopy(vw.get(), clusters);
- writer.append(new IntWritable(closest.getId()), new WeightedVectorWritable(1, vw.get()));
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
+ for (VectorWritable value : new SequenceFileValueIterable<VectorWritable>(s.getPath(), conf)) {
+ Canopy closest = clusterer.findClosestCanopy(value.get(), clusters);
+ writer.append(new IntWritable(closest.getId()), new WeightedVectorWritable(1, value.get()));
}
} finally {
- reader.close();
writer.close();
}
}
@@ -349,7 +331,7 @@ public class CanopyDriver extends Abstra
FileInputFormat.addInputPath(job, points);
Path outPath = new Path(output, DEFAULT_CLUSTERED_POINTS_DIRECTORY);
FileOutputFormat.setOutputPath(job, outPath);
- HadoopUtil.overwriteOutput(outPath);
+ HadoopUtil.delete(conf, outPath);
if (!job.waitForCompletion(true)) {
throw new InterruptedException("Canopy Clustering failed processing " + canopies.toString());
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/ClusterMapper.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/ClusterMapper.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/ClusterMapper.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/canopy/ClusterMapper.java Thu Mar 31 09:25:25 2011
@@ -26,14 +26,12 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.clustering.WeightedVectorWritable;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
@@ -42,7 +40,8 @@ public class ClusterMapper extends Mappe
private CanopyClusterer canopyClusterer;
@Override
- protected void map(WritableComparable<?> key, VectorWritable point, Context context) throws IOException, InterruptedException {
+ protected void map(WritableComparable<?> key, VectorWritable point, Context context)
+ throws IOException, InterruptedException {
canopyClusterer.emitPointToClosestCanopy(point.get(), canopies, context);
}
@@ -50,9 +49,6 @@ public class ClusterMapper extends Mappe
/**
* Configure the mapper by providing its canopies. Used by unit tests.
- *
- * @param canopies
- * a List<Canopy>
*/
public void config(Collection<Canopy> canopies) {
this.canopies.clear();
@@ -69,32 +65,14 @@ public class ClusterMapper extends Mappe
String clustersIn = conf.get(CanopyConfigKeys.CANOPY_PATH_KEY);
// filter out the files
- PathFilter clusterFileFilter = new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return path.getName().startsWith("part");
- }
- };
- if ((clustersIn != null) && (clustersIn.length() > 0)) {
- try {
- Path clusterPath = new Path(clustersIn,"*");
- FileSystem fs = clusterPath.getFileSystem(conf);
- FileStatus[] files = fs.listStatus(FileUtil.stat2Paths(fs.globStatus(clusterPath, clusterFileFilter)), clusterFileFilter);
- for (FileStatus file : files) {
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, file.getPath(), conf);
- try {
- Writable key = new Text();
- Canopy value = new Canopy();
- while (reader.next(key, value)) {
- canopies.add(value);
- value = new Canopy();
- }
- } finally {
- reader.close();
- }
+ if (clustersIn != null && clustersIn.length() > 0) {
+ Path clusterPath = new Path(clustersIn,"*");
+ FileSystem fs = clusterPath.getFileSystem(conf);
+ Path[] paths = FileUtil.stat2Paths(fs.globStatus(clusterPath, PathFilters.partFilter()));
+ for (FileStatus file : fs.listStatus(paths, PathFilters.partFilter())) {
+ for (Canopy value : new SequenceFileValueIterable<Canopy>(file.getPath(), conf)) {
+ canopies.add(value);
}
- } catch (IOException e) {
- throw new IllegalStateException(e);
}
if (canopies.isEmpty()) {
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=1087225&r1=1087224&r2=1087225&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 Mar 31 09:25:25 2011
@@ -22,27 +22,26 @@ 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;
import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.mahout.clustering.WeightedVectorWritable;
-import org.apache.mahout.clustering.kmeans.OutputLogFilter;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable;
import org.apache.mahout.math.VectorWritable;
-public class DirichletClusterMapper extends Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable> {
+public class DirichletClusterMapper
+ extends Mapper<WritableComparable<?>, VectorWritable, IntWritable, WeightedVectorWritable> {
private List<DirichletCluster> clusters;
private DirichletClusterer clusterer;
@Override
- protected void map(WritableComparable<?> key, VectorWritable vector, Context context) throws IOException, InterruptedException {
+ protected void map(WritableComparable<?> key, VectorWritable vector, Context context)
+ throws IOException, InterruptedException {
clusterer.emitPointToClusters(vector, clusters, context);
}
@@ -50,19 +49,13 @@ public class DirichletClusterMapper exte
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
Configuration conf = context.getConfiguration();
- try {
- clusters = getClusters(conf);
- for (DirichletCluster cluster : clusters) {
- cluster.getModel().configure(conf);
- }
- String emitMostLikely = conf.get(DirichletDriver.EMIT_MOST_LIKELY_KEY);
- String threshold = conf.get(DirichletDriver.THRESHOLD_KEY);
- clusterer = new DirichletClusterer(Boolean.parseBoolean(emitMostLikely), Double.parseDouble(threshold));
- } catch (SecurityException e) {
- throw new IllegalStateException(e);
- } catch (IllegalArgumentException e) {
- throw new IllegalStateException(e);
+ clusters = getClusters(conf);
+ for (DirichletCluster cluster : clusters) {
+ cluster.getModel().configure(conf);
}
+ String emitMostLikely = conf.get(DirichletDriver.EMIT_MOST_LIKELY_KEY);
+ String threshold = conf.get(DirichletDriver.THRESHOLD_KEY);
+ clusterer = new DirichletClusterer(Boolean.parseBoolean(emitMostLikely), Double.parseDouble(threshold));
}
public static List<DirichletCluster> getClusters(Configuration conf) {
@@ -72,26 +65,12 @@ public class DirichletClusterMapper exte
protected static List<DirichletCluster> loadClusters(Configuration conf, Path stateIn) {
List<DirichletCluster> clusters = new ArrayList<DirichletCluster>();
- try {
- 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(), conf);
- try {
- Writable key = new Text();
- DirichletCluster cluster = new DirichletCluster();
- while (reader.next(key, cluster)) {
- clusters.add(cluster);
- cluster = new DirichletCluster();
- }
- } finally {
- reader.close();
- }
- }
- return clusters;
- } catch (IOException e) {
- throw new IllegalStateException(e);
+ for (DirichletCluster value :
+ new SequenceFileDirValueIterable<DirichletCluster>(
+ stateIn, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ clusters.add(value);
}
+ return clusters;
}
}
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=1087225&r1=1087224&r2=1087225&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 Mar 31 09:25:25 2011
@@ -29,7 +29,6 @@ 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.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -42,11 +41,14 @@ import org.apache.mahout.clustering.Weig
import org.apache.mahout.clustering.dirichlet.models.AbstractVectorModelDistribution;
import org.apache.mahout.clustering.dirichlet.models.DistanceMeasureClusterDistribution;
import org.apache.mahout.clustering.dirichlet.models.NormalModelDistribution;
-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.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
@@ -72,9 +74,7 @@ public class DirichletDriver extends Abs
}
@Override
- public int run(String[] args)
- throws IOException, ClassNotFoundException, InstantiationException, IllegalAccessException,
- NoSuchMethodException, InvocationTargetException, InterruptedException {
+ public int run(String[] args) throws Exception {
addInputOption();
addOutputOption();
addOption(DefaultOptionCreator.maxIterationsOption().create());
@@ -102,7 +102,7 @@ public class DirichletDriver extends Abs
Path input = getInputPath();
Path output = getOutputPath();
if (hasOption(DefaultOptionCreator.OVERWRITE_OPTION)) {
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(getConf(), output);
}
String modelFactory = getOption(MODEL_DISTRIBUTION_CLASS_OPTION);
String modelPrototype = getOption(MODEL_PROTOTYPE_CLASS_OPTION);
@@ -171,7 +171,7 @@ public class DirichletDriver extends Abs
boolean emitMostLikely,
double threshold,
boolean runSequential)
- throws IOException, InstantiationException, ClassNotFoundException, InterruptedException, IllegalAccessException {
+ throws IOException, ClassNotFoundException, InterruptedException {
Path clustersOut =
buildClusters(conf, input, output, modelDistribution, numModels, maxIterations, alpha0, runSequential);
if (runClustering) {
@@ -220,7 +220,7 @@ public class DirichletDriver extends Abs
boolean emitMostLikely,
double threshold,
boolean runSequential)
- throws IOException, InstantiationException, IllegalAccessException, ClassNotFoundException, InterruptedException {
+ throws IOException, ClassNotFoundException, InterruptedException {
run(new Configuration(),
input,
output,
@@ -276,20 +276,16 @@ public class DirichletDriver extends Abs
/**
* Read the first input vector to determine the prototype size for the modelPrototype
*/
- public static int readPrototypeSize(Path input) throws IOException, InstantiationException, IllegalAccessException {
+ public static int readPrototypeSize(Path input) throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(input.toUri(), conf);
- FileStatus[] status = fs.listStatus(input, new OutputLogFilter());
+ FileStatus[] status = fs.listStatus(input, PathFilters.logsCRCFilter());
int protoSize = 0;
if (status.length > 0) {
FileStatus s = status[0];
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, s.getPath(), conf);
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable value = new VectorWritable();
- if (reader.next(key, value)) {
+ for (VectorWritable value : new SequenceFileValueIterable<VectorWritable>(s.getPath(), true, conf)) {
protoSize = value.get().size();
}
- reader.close();
}
return protoSize;
}
@@ -391,7 +387,7 @@ public class DirichletDriver extends Abs
int maxIterations,
double alpha0,
boolean runSequential)
- throws IOException, InstantiationException, ClassNotFoundException, InterruptedException, IllegalAccessException {
+ throws IOException, ClassNotFoundException, InterruptedException {
Path clustersIn = new Path(output, Cluster.INITIAL_CLUSTERS_DIR);
writeInitialState(output, clustersIn, modelDistribution, numClusters, alpha0);
@@ -410,8 +406,7 @@ public class DirichletDriver extends Abs
int numClusters,
int maxIterations,
double alpha0,
- Path clustersIn)
- throws IOException, InstantiationException, IllegalAccessException {
+ Path clustersIn) throws IOException {
for (int iteration = 1; iteration <= maxIterations; iteration++) {
log.info("Iteration {}", iteration);
// point the output to a new directory per iteration
@@ -431,20 +426,10 @@ public class DirichletDriver extends Abs
newModel.configure(conf);
}
DirichletClusterer clusterer = new DirichletClusterer(state);
- 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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.observe(newModels, vw);
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- }
- } finally {
- reader.close();
- }
+ for (VectorWritable value :
+ new SequenceFileDirValueIterable<VectorWritable>(
+ input, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ clusterer.observe(newModels, value);
}
clusterer.updateModels(newModels);
writeState(output, clustersOut, numClusters, state);
@@ -498,7 +483,7 @@ public class DirichletDriver extends Abs
boolean emitMostLikely,
double threshold,
boolean runSequential)
- throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
+ throws IOException, InterruptedException, ClassNotFoundException {
if (runSequential) {
clusterDataSeq(conf, input, stateIn, output, emitMostLikely, threshold);
} else {
@@ -506,37 +491,35 @@ public class DirichletDriver extends Abs
}
}
- private static void clusterDataSeq(Configuration conf, Path input, Path stateIn, Path output, boolean emitMostLikely, double threshold)
- throws IOException, InstantiationException, IllegalAccessException {
+ private static void clusterDataSeq(Configuration conf,
+ Path input,
+ Path stateIn,
+ Path output,
+ boolean emitMostLikely,
+ double threshold) throws IOException {
+
List<DirichletCluster> clusters = DirichletClusterMapper.loadClusters(conf, stateIn);
-
- for(int i=0; i<clusters.size(); i++)
- {
- Cluster cluster = clusters.get(i).getModel();
- cluster.configure(conf);
+
+ for (DirichletCluster cluster : clusters) {
+ cluster.getModel().configure(conf);
}
DirichletClusterer clusterer = new DirichletClusterer(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());
+ FileStatus[] status = fs.listStatus(input, PathFilters.logsCRCFilter());
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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.emitPointToClusters(vw, clusters, writer);
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
+ for (VectorWritable value : new SequenceFileValueIterable<VectorWritable>(s.getPath(), conf)) {
+ clusterer.emitPointToClusters(value, clusters, writer);
}
} finally {
- reader.close();
writer.close();
}
}
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=1087225&r1=1087224&r2=1087225&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 Mar 31 09:25:25 2011
@@ -20,10 +20,7 @@ package org.apache.mahout.clustering.dir
import java.io.IOException;
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.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
@@ -31,7 +28,10 @@ import org.apache.hadoop.mapreduce.Mappe
import org.apache.mahout.clustering.JsonModelDistributionAdapter;
import org.apache.mahout.clustering.ModelDistribution;
import org.apache.mahout.clustering.dirichlet.models.AbstractVectorModelDistribution;
-import org.apache.mahout.clustering.kmeans.OutputLogFilter;
+import org.apache.mahout.common.Pair;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirIterable;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.VectorWritable;
@@ -43,7 +43,8 @@ public class DirichletMapper extends Map
private DirichletClusterer clusterer;
@Override
- protected void map(WritableComparable<?> key, VectorWritable v, Context context) throws IOException, InterruptedException {
+ protected void map(WritableComparable<?> key, VectorWritable v, Context context)
+ throws IOException, InterruptedException {
int k = clusterer.assignToModel(v);
context.write(new Text(String.valueOf(k)), v);
}
@@ -51,23 +52,15 @@ public class DirichletMapper extends Map
@Override
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
- try {
- DirichletState dirichletState = getDirichletState(context.getConfiguration());
- for (DirichletCluster cluster : dirichletState.getClusters()) {
- cluster.getModel().configure(context.getConfiguration());
- }
- clusterer = new DirichletClusterer(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) {
- throw new IllegalStateException(e);
- } catch (IllegalArgumentException e) {
- throw new IllegalStateException(e);
+ DirichletState dirichletState = getDirichletState(context.getConfiguration());
+ for (DirichletCluster cluster : dirichletState.getClusters()) {
+ cluster.getModel().configure(context.getConfiguration());
+ }
+ clusterer = new DirichletClusterer(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)));
}
}
@@ -86,35 +79,21 @@ public class DirichletMapper extends Map
String numClusters = conf.get(DirichletDriver.NUM_CLUSTERS_KEY);
String alpha0 = conf.get(DirichletDriver.ALPHA_0_KEY);
- try {
- return loadState(conf, statePath, modelDistribution, Double.parseDouble(alpha0), Integer.parseInt(numClusters));
- } catch (IOException e) {
- throw new IllegalStateException(e);
- }
+ return loadState(conf, statePath, modelDistribution, Double.parseDouble(alpha0), Integer.parseInt(numClusters));
}
protected static DirichletState loadState(Configuration conf,
String statePath,
ModelDistribution<VectorWritable> modelDistribution,
double alpha,
- int k) throws IOException {
+ int k) {
DirichletState state = DirichletDriver.createState(modelDistribution, 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 {
- Writable key = new Text();
- DirichletCluster cluster = new DirichletCluster();
- while (reader.next(key, cluster)) {
- int index = Integer.parseInt(key.toString());
- state.getClusters().set(index, cluster);
- cluster = new DirichletCluster();
- }
- } finally {
- reader.close();
- }
+ for (Pair<Writable,DirichletCluster> record :
+ new SequenceFileDirIterable<Writable,DirichletCluster>(
+ path, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ int index = Integer.parseInt(record.getFirst().toString());
+ state.getClusters().set(index, record.getSecond());
}
// TODO: with more than one mapper, they will all have different mixtures. Will this matter?
state.setMixture(UncommonDistributions.rDirichlet(state.totalCounts(), alpha));
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=1087225&r1=1087224&r2=1087225&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 Mar 31 09:25:25 2011
@@ -38,23 +38,15 @@ public class DirichletReducer extends Re
@Override
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
- try {
- DirichletState state = DirichletMapper.getDirichletState(context.getConfiguration());
- clusterer = new DirichletClusterer(state);
- List<DirichletCluster> oldModels = state.getClusters();
- for (DirichletCluster cluster : oldModels) {
- cluster.getModel().configure(context.getConfiguration());
- }
- this.newModels = (Cluster[]) state.getModelFactory().sampleFromPosterior(state.getModels());
- for (Cluster cluster : newModels) {
- cluster.configure(context.getConfiguration());
- }
- } catch (NumberFormatException e) {
- throw new IllegalStateException(e);
- } catch (SecurityException e) {
- throw new IllegalStateException(e);
- } catch (IllegalArgumentException e) {
- throw new IllegalStateException(e);
+ DirichletState state = DirichletMapper.getDirichletState(context.getConfiguration());
+ clusterer = new DirichletClusterer(state);
+ List<DirichletCluster> oldModels = state.getClusters();
+ for (DirichletCluster cluster : oldModels) {
+ cluster.getModel().configure(context.getConfiguration());
+ }
+ this.newModels = (Cluster[]) state.getModelFactory().sampleFromPosterior(state.getModels());
+ for (Cluster cluster : newModels) {
+ cluster.configure(context.getConfiguration());
}
}
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=1087225&r1=1087224&r2=1087225&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 Mar 31 09:25:25 2011
@@ -27,11 +27,9 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -42,13 +40,17 @@ import org.apache.mahout.clustering.Abst
import org.apache.mahout.clustering.Cluster;
import org.apache.mahout.clustering.ClusterObservations;
import org.apache.mahout.clustering.WeightedVectorWritable;
-import org.apache.mahout.clustering.kmeans.OutputLogFilter;
import org.apache.mahout.clustering.kmeans.RandomSeedGenerator;
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.common.distance.SquaredEuclideanDistanceMeasure;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterator;
import org.apache.mahout.math.VectorWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -102,7 +104,7 @@ public class FuzzyKMeansDriver extends A
int maxIterations = Integer.parseInt(getOption(DefaultOptionCreator.MAX_ITERATIONS_OPTION));
if (hasOption(DefaultOptionCreator.OVERWRITE_OPTION)) {
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(getConf(), output);
}
boolean emitMostLikely = Boolean.parseBoolean(getOption(DefaultOptionCreator.EMIT_MOST_LIKELY_OPTION));
double threshold = Double.parseDouble(getOption(DefaultOptionCreator.THRESHOLD_OPTION));
@@ -166,8 +168,7 @@ public class FuzzyKMeansDriver extends A
boolean runClustering,
boolean emitMostLikely,
double threshold,
- boolean runSequential) throws IOException, ClassNotFoundException, InterruptedException,
- InstantiationException, IllegalAccessException {
+ boolean runSequential) throws IOException, ClassNotFoundException, InterruptedException {
Path clustersOut = buildClusters(new Configuration(),
input,
clustersIn,
@@ -286,11 +287,11 @@ public class FuzzyKMeansDriver extends A
boolean emitMostLikely,
double threshold,
boolean runSequential)
- throws IOException, ClassNotFoundException, InterruptedException, InstantiationException, IllegalAccessException {
+ throws IOException, ClassNotFoundException, InterruptedException {
Path clustersOut =
buildClusters(conf, input, clustersIn, output, measure, convergenceDelta, maxIterations, m, runSequential);
if (runClustering) {
- log.info("Clustering ");
+ log.info("Clustering");
clusterData(input,
clustersOut,
new Path(output, Cluster.CLUSTERED_POINTS_DIR),
@@ -305,7 +306,6 @@ public class FuzzyKMeansDriver extends A
/**
* Iterate over the input vectors to produce cluster directories for each iteration
- * @param conf TODO
* @param input
* the directory pathname for input points
* @param clustersIn
@@ -324,8 +324,6 @@ public class FuzzyKMeansDriver extends A
* @param runSequential if true run in sequential execution mode
*
* @return the Path of the final clusters directory
- * @throws ClassNotFoundException
- * @throws InterruptedException
*/
public static Path buildClusters(Configuration conf,
Path input,
@@ -336,7 +334,7 @@ public class FuzzyKMeansDriver extends A
int maxIterations,
float m,
boolean runSequential)
- throws IOException, InstantiationException, IllegalAccessException, InterruptedException, ClassNotFoundException {
+ throws IOException, InterruptedException, ClassNotFoundException {
if (runSequential) {
return buildClustersSeq(input, clustersIn, output, measure, convergenceDelta, maxIterations, m);
} else {
@@ -350,7 +348,7 @@ public class FuzzyKMeansDriver extends A
DistanceMeasure measure,
double convergenceDelta,
int maxIterations,
- float m) throws IOException, InstantiationException, IllegalAccessException {
+ float m) throws IOException {
FuzzyKMeansClusterer clusterer = new FuzzyKMeansClusterer(measure, convergenceDelta, m);
List<SoftCluster> clusters = new ArrayList<SoftCluster>();
@@ -364,19 +362,10 @@ public class FuzzyKMeansDriver extends A
log.info("Fuzzy k-Means Iteration: " + iteration);
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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.addPointToClusters(clusters, vw.get());
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- }
- } finally {
- reader.close();
- }
+ for (VectorWritable value :
+ new SequenceFileDirValueIterable<VectorWritable>(
+ input, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ clusterer.addPointToClusters(clusters,value.get());
}
converged = clusterer.testConvergence(clusters);
Path clustersOut = new Path(output, Cluster.CLUSTERS_DIR + iteration);
@@ -387,9 +376,13 @@ public class FuzzyKMeansDriver extends A
SoftCluster.class);
try {
for (SoftCluster cluster : clusters) {
- log.debug("Writing Cluster:{} center:{} numPoints:{} radius:{} to: {}", new Object[] { cluster.getId(),
- AbstractCluster.formatVector(cluster.getCenter(), null), cluster.getNumPoints(),
- AbstractCluster.formatVector(cluster.getRadius(), null), clustersOut.getName() });
+ log.debug("Writing Cluster:{} center:{} numPoints:{} radius:{} to: {}",
+ new Object[] { cluster.getId(),
+ AbstractCluster.formatVector(cluster.getCenter(), null),
+ cluster.getNumPoints(),
+ AbstractCluster.formatVector(cluster.getRadius(), null),
+ clustersOut.getName()
+ });
writer.append(new Text(cluster.getIdentifier()), cluster);
}
} finally {
@@ -455,7 +448,7 @@ public class FuzzyKMeansDriver extends A
boolean emitMostLikely,
double threshold,
boolean runSequential)
- throws IOException, ClassNotFoundException, InterruptedException, InstantiationException, IllegalAccessException {
+ throws IOException, ClassNotFoundException, InterruptedException {
if (runSequential) {
clusterDataSeq(input, clustersIn, output, measure, convergenceDelta, m);
} else {
@@ -468,7 +461,7 @@ public class FuzzyKMeansDriver extends A
Path output,
DistanceMeasure measure,
double convergenceDelta,
- float m) throws IOException, InstantiationException, IllegalAccessException {
+ float m) throws IOException {
FuzzyKMeansClusterer clusterer = new FuzzyKMeansClusterer(measure, convergenceDelta, m);
List<SoftCluster> clusters = new ArrayList<SoftCluster>();
FuzzyKMeansUtil.configureWithClusterInfo(clustersIn, clusters);
@@ -477,24 +470,19 @@ public class FuzzyKMeansDriver extends A
}
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(input.toUri(), conf);
- FileStatus[] status = fs.listStatus(input, new OutputLogFilter());
+ FileStatus[] status = fs.listStatus(input, PathFilters.logsCRCFilter());
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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.emitPointToClusters(vw, clusters, writer);
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
+ for (VectorWritable value : new SequenceFileValueIterable<VectorWritable>(s.getPath(), conf)) {
+ clusterer.emitPointToClusters(value, clusters, writer);
}
} finally {
- reader.close();
writer.close();
}
}
@@ -553,41 +541,22 @@ public class FuzzyKMeansDriver extends A
Path clusterPath = new Path(filePath, "*");
Collection<Path> result = new ArrayList<Path>();
- PathFilter clusterFileFilter = new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return path.getName().startsWith("part");
- }
- };
-
FileStatus[] matches =
- fs.listStatus(FileUtil.stat2Paths(fs.globStatus(clusterPath, clusterFileFilter)), clusterFileFilter);
+ fs.listStatus(FileUtil.stat2Paths(fs.globStatus(clusterPath, PathFilters.partFilter())),
+ PathFilters.partFilter());
for (FileStatus match : matches) {
result.add(fs.makeQualified(match.getPath()));
}
boolean converged = true;
- for (Path p : result) {
-
- SequenceFile.Reader reader = null;
-
- try {
- reader = new SequenceFile.Reader(fs, p, conf);
- /*
- * new KeyValueLineRecordReader(conf, new FileSplit(p, 0, fs .getFileStatus(p).getLen(), (String[])
- * null));
- */
- Writable key = new Text();
- SoftCluster value = new SoftCluster();
- while (converged && reader.next(key, value)) {
- converged = value.isConverged();
- }
- } finally {
- if (reader != null) {
- reader.close();
- }
+ for (Path path : result) {
+ SequenceFileValueIterator<SoftCluster> iterator =
+ new SequenceFileValueIterator<SoftCluster>(path, true, conf);
+ while (converged && iterator.hasNext()) {
+ converged = iterator.next().isConverged();
}
+ iterator.close();
}
return converged;
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansUtil.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansUtil.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansUtil.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/fuzzykmeans/FuzzyKMeansUtil.java Thu Mar 31 09:25:25 2011
@@ -26,95 +26,55 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
-import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Writable;
import org.apache.mahout.clustering.canopy.Canopy;
import org.apache.mahout.clustering.kmeans.Cluster;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
final class FuzzyKMeansUtil {
- private static final Logger log = LoggerFactory.getLogger(FuzzyKMeansUtil.class);
private FuzzyKMeansUtil() {
}
/** Configure the mapper with the cluster info */
- public static void configureWithClusterInfo(Path clusterPathStr, Collection<SoftCluster> clusters) {
+ public static void configureWithClusterInfo(Path clusterPathStr, Collection<SoftCluster> clusters)
+ throws IOException {
// Get the path location where the cluster Info is stored
- Configuration job = new Configuration();
+ Configuration conf = new Configuration();
Path clusterPath = new Path(clusterPathStr, "*");
Collection<Path> result = new ArrayList<Path>();
- // filter out the files
- PathFilter clusterFileFilter = new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return path.getName().startsWith("part");
- }
- };
- try {
- // get all filtered file names in result list
- FileSystem fs = clusterPath.getFileSystem(job);
- FileStatus[] matches = fs.listStatus(FileUtil.stat2Paths(
- fs.globStatus(clusterPath, clusterFileFilter)), clusterFileFilter);
+ // get all filtered file names in result list
+ FileSystem fs = clusterPath.getFileSystem(conf);
+ FileStatus[] matches = fs.listStatus(FileUtil.stat2Paths(fs.globStatus(clusterPath, PathFilters.partFilter())),
+ PathFilters.partFilter());
- for (FileStatus match : matches) {
- result.add(fs.makeQualified(match.getPath()));
- }
+ for (FileStatus match : matches) {
+ result.add(fs.makeQualified(match.getPath()));
+ }
- // iterate thru the result path list
- for (Path path : result) {
- // RecordReader<Text, Text> recordReader = null;
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, job);
- try {
- // recordReader = new KeyValueLineRecordReader(job, new FileSplit(path, 0,
- // fs.getFileStatus(path).getLen(), (String[]) null));
- Class<?> valueClass = reader.getValueClass();
- Writable key;
- try {
- key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- } catch (InstantiationException e) { // Should not be possible
- log.error("Exception", e);
- throw new IllegalStateException(e);
- } catch (IllegalAccessException e) {
- log.error("Exception", e);
- throw new IllegalStateException(e);
- }
- if (valueClass.equals(Cluster.class)) {
- Cluster value = new Cluster();
- while (reader.next(key, value)) {
- // get the cluster info
- SoftCluster theCluster = new SoftCluster(value.getCenter(), value.getId(), value.getMeasure());
- clusters.add(theCluster);
- value = new Cluster();
- }
- } else if (valueClass.equals(SoftCluster.class)) {
- SoftCluster value = new SoftCluster();
- while (reader.next(key, value)) {
- // get the cluster info
- clusters.add(value);
- value = new SoftCluster();
- }
- } else if (valueClass.equals(Canopy.class)) {
- Canopy value = new Canopy();
- while (reader.next(key, value)) {
- // get the cluster info
- SoftCluster theCluster = new SoftCluster(value.getCenter(), value.getId(), value.getMeasure());
- clusters.add(theCluster);
- value = new Canopy();
- }
- }
- } finally {
- reader.close();
+ // iterate thru the result path list
+ for (Path path : result) {
+ for (Writable value : new SequenceFileValueIterable<Writable>(path, conf)) {
+ Class<? extends Writable> valueClass = value.getClass();
+ if (valueClass.equals(Cluster.class)) {
+ // get the cluster info
+ Cluster cluster = (Cluster) value;
+ clusters.add(new SoftCluster(cluster.getCenter(), cluster.getId(), cluster.getMeasure()));
+ } else if (valueClass.equals(SoftCluster.class)) {
+ // get the cluster info
+ clusters.add((SoftCluster) value);
+ } else if (valueClass.equals(Canopy.class)) {
+ // get the cluster info
+ Canopy canopy = (Canopy) value;
+ clusters.add(new SoftCluster(canopy.getCenter(), canopy.getId(), canopy.getMeasure()));
+ } else {
+ throw new IllegalStateException("Bad value class: " + valueClass);
}
}
-
- } catch (IOException e) {
- log.info("Exception occurred in loading clusters:", e);
- throw new IllegalStateException(e);
}
+
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansDriver.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansDriver.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansDriver.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansDriver.java Thu Mar 31 09:25:25 2011
@@ -27,7 +27,6 @@ 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.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -42,6 +41,11 @@ import org.apache.mahout.common.HadoopUt
import org.apache.mahout.common.commandline.DefaultOptionCreator;
import org.apache.mahout.common.distance.DistanceMeasure;
import org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.PathType;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterator;
import org.apache.mahout.math.VectorWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,7 +92,7 @@ public class KMeansDriver extends Abstra
double convergenceDelta = Double.parseDouble(getOption(DefaultOptionCreator.CONVERGENCE_DELTA_OPTION));
int maxIterations = Integer.parseInt(getOption(DefaultOptionCreator.MAX_ITERATIONS_OPTION));
if (hasOption(DefaultOptionCreator.OVERWRITE_OPTION)) {
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(getConf(), output);
}
ClassLoader ccl = Thread.currentThread().getContextClassLoader();
DistanceMeasure measure = ccl.loadClass(measureClass).asSubclass(DistanceMeasure.class).newInstance();
@@ -132,7 +136,7 @@ public class KMeansDriver extends Abstra
int maxIterations,
boolean runClustering,
boolean runSequential)
- throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
+ throws IOException, InterruptedException, ClassNotFoundException {
// iterate until the clusters converge
String delta = Double.toString(convergenceDelta);
@@ -183,7 +187,7 @@ public class KMeansDriver extends Abstra
int maxIterations,
boolean runClustering,
boolean runSequential)
- throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
+ throws IOException, InterruptedException, ClassNotFoundException {
run(new Configuration(),
input,
clustersIn,
@@ -223,7 +227,7 @@ public class KMeansDriver extends Abstra
int maxIterations,
String delta,
boolean runSequential)
- throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
+ throws IOException, InterruptedException, ClassNotFoundException {
if (runSequential) {
return buildClustersSeq(input, clustersIn, output, measure, maxIterations, delta);
} else {
@@ -237,7 +241,7 @@ public class KMeansDriver extends Abstra
DistanceMeasure measure,
int maxIterations,
String delta)
- throws InstantiationException, IllegalAccessException, IOException {
+ throws IOException {
KMeansClusterer clusterer = new KMeansClusterer(measure);
Collection<Cluster> clusters = new ArrayList<Cluster>();
@@ -252,19 +256,10 @@ public class KMeansDriver extends Abstra
log.info("K-Means Iteration: " + iteration);
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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.addPointToNearestCluster(vw.get(), clusters);
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- }
- } finally {
- reader.close();
- }
+ for (VectorWritable value :
+ new SequenceFileDirValueIterable<VectorWritable>(
+ input, PathType.LIST, PathFilters.logsCRCFilter(), conf)) {
+ clusterer.addPointToNearestCluster(value.get(), clusters);
}
converged = clusterer.testConvergence(clusters, Double.parseDouble(delta));
Path clustersOut = new Path(output, AbstractCluster.CLUSTERS_DIR + iteration);
@@ -313,7 +308,6 @@ public class KMeansDriver extends Abstra
/**
* Run the job using supplied arguments
- * @param conf TODO
* @param input
* the directory pathname for input points
* @param clustersIn
@@ -355,7 +349,7 @@ public class KMeansDriver extends Abstra
FileOutputFormat.setOutputPath(job, clustersOut);
job.setJarByClass(KMeansDriver.class);
- HadoopUtil.overwriteOutput(clustersOut);
+ HadoopUtil.delete(conf, clustersOut);
if (!job.waitForCompletion(true)) {
throw new InterruptedException("K-Means Iteration failed processing " + clustersIn.toString());
}
@@ -374,27 +368,13 @@ public class KMeansDriver extends Abstra
* if there was an IO error
*/
private static boolean isConverged(Path filePath, Configuration conf, FileSystem fs) throws IOException {
- FileStatus[] parts = fs.listStatus(filePath);
- for (FileStatus part : parts) {
- String name = part.getPath().getName();
- if (name.startsWith("part") && !name.endsWith(".crc")) {
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, part.getPath(), conf);
- try {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- Cluster value = new Cluster();
- while (reader.next(key, value)) {
- if (!value.isConverged()) {
- return false;
- }
- }
- } catch (InstantiationException e) { // shouldn't happen
- log.error("Exception", e);
- throw new IllegalStateException(e);
- } catch (IllegalAccessException e) {
- log.error("Exception", e);
- throw new IllegalStateException(e);
- } finally {
- reader.close();
+ for (FileStatus part : fs.listStatus(filePath, PathFilters.partFilter())) {
+ SequenceFileValueIterator<Cluster> iterator = new SequenceFileValueIterator<Cluster>(part.getPath(), true, conf);
+ while (iterator.hasNext()) {
+ Cluster value = iterator.next();
+ if (!value.isConverged()) {
+ iterator.close();
+ return false;
}
}
}
@@ -403,7 +383,6 @@ public class KMeansDriver extends Abstra
/**
* Run the job using supplied arguments
- * @param conf TODO
* @param input
* the directory pathname for input points
* @param clustersIn
@@ -423,7 +402,7 @@ public class KMeansDriver extends Abstra
DistanceMeasure measure,
String convergenceDelta,
boolean runSequential)
- throws IOException, InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException {
+ throws IOException, InterruptedException, ClassNotFoundException {
if (log.isInfoEnabled()) {
log.info("Running Clustering");
@@ -438,7 +417,7 @@ public class KMeansDriver extends Abstra
}
private static void clusterDataSeq(Path input, Path clustersIn, Path output, DistanceMeasure measure)
- throws IOException, InstantiationException, IllegalAccessException {
+ throws IOException {
KMeansClusterer clusterer = new KMeansClusterer(measure);
Collection<Cluster> clusters = new ArrayList<Cluster>();
@@ -448,24 +427,19 @@ public class KMeansDriver extends Abstra
}
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(input.toUri(), conf);
- FileStatus[] status = fs.listStatus(input, new OutputLogFilter());
+ FileStatus[] status = fs.listStatus(input, PathFilters.logsCRCFilter());
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 {
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, vw)) {
- clusterer.emitPointToNearestCluster(vw.get(), clusters, writer);
- vw = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
+ for (VectorWritable value : new SequenceFileValueIterable<VectorWritable>(s.getPath(), conf)) {
+ clusterer.emitPointToNearestCluster(value.get(), clusters, writer);
}
} finally {
- reader.close();
writer.close();
}
}
@@ -491,7 +465,7 @@ public class KMeansDriver extends Abstra
job.setOutputValueClass(WeightedVectorWritable.class);
FileInputFormat.setInputPaths(job, input);
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(conf, output);
FileOutputFormat.setOutputPath(job, output);
job.setMapperClass(KMeansClusterMapper.class);
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansUtil.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansUtil.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansUtil.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/KMeansUtil.java Thu Mar 31 09:25:25 2011
@@ -26,86 +26,48 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
-import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Writable;
import org.apache.mahout.clustering.canopy.Canopy;
-import org.apache.mahout.common.IOUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterable;
final class KMeansUtil {
- private static final Logger log = LoggerFactory.getLogger(KMeansUtil.class);
-
private KMeansUtil() {
}
/** Configure the mapper with the cluster info */
- public static void configureWithClusterInfo(Path clusterPathStr, Collection<Cluster> clusters) {
+ public static void configureWithClusterInfo(Path clusterPathStr, Collection<Cluster> clusters) throws IOException {
// Get the path location where the cluster Info is stored
Configuration conf = new Configuration();
Path clusterPath = new Path(clusterPathStr, "*");
Collection<Path> result = new ArrayList<Path>();
- // filter out the files
- PathFilter clusterFileFilter = new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return path.getName().startsWith("part");
- }
- };
-
- try {
- // get all filtered file names in result list
- FileSystem fs = clusterPath.getFileSystem(conf);
- FileStatus[] matches = fs.listStatus(FileUtil.stat2Paths(fs.globStatus(clusterPath, clusterFileFilter)), clusterFileFilter);
+ // get all filtered file names in result list
+ FileSystem fs = clusterPath.getFileSystem(conf);
+ FileStatus[] matches = fs.listStatus(FileUtil.stat2Paths(fs.globStatus(clusterPath, PathFilters.partFilter())),
+ PathFilters.partFilter());
- for (FileStatus match : matches) {
- result.add(fs.makeQualified(match.getPath()));
- }
+ for (FileStatus match : matches) {
+ result.add(fs.makeQualified(match.getPath()));
+ }
- // iterate thru the result path list
- for (Path path : result) {
- SequenceFile.Reader reader = null;
- try {
- reader = new SequenceFile.Reader(fs, path, conf);
- Class<?> valueClass = reader.getValueClass();
- Writable key;
- try {
- key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- } catch (InstantiationException e) { // Should not be possible
- log.error("Exception", e);
- throw new IllegalStateException(e);
- } catch (IllegalAccessException e) {
- log.error("Exception", e);
- throw new IllegalStateException(e);
- }
- if (valueClass.equals(Cluster.class)) {
- Cluster value = new Cluster();
- while (reader.next(key, value)) {
- // get the cluster info
- clusters.add(value);
- value = new Cluster();
- }
- } else if (valueClass.equals(Canopy.class)) {
- Canopy value = new Canopy();
- while (reader.next(key, value)) {
- // get the cluster info
- Cluster cluster = new Cluster(value.getCenter(), value.getId(), value.getMeasure());
- clusters.add(cluster);
- value = new Canopy();
- }
- }
- } finally {
- IOUtils.quietClose(reader);
+ // iterate thru the result path list
+ for (Path path : result) {
+ for (Writable value : new SequenceFileValueIterable<Writable>(path, conf)) {
+ Class<? extends Writable> valueClass = value.getClass();
+ if (valueClass.equals(Cluster.class)) {
+ // get the cluster info
+ clusters.add((Cluster) value);
+ } else if (valueClass.equals(Canopy.class)) {
+ // get the cluster info
+ Canopy canopy = (Canopy) value;
+ clusters.add(new Cluster(canopy.getCenter(), canopy.getId(), canopy.getMeasure()));
+ } else {
+ throw new IllegalStateException("Bad value class: " + valueClass);
}
}
-
- } catch (IOException e) {
- log.info("Exception occurred in loading clusters:", e);
- throw new IllegalStateException(e);
}
}
Modified: mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/RandomSeedGenerator.java
URL: http://svn.apache.org/viewvc/mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/RandomSeedGenerator.java?rev=1087225&r1=1087224&r2=1087225&view=diff
==============================================================================
--- mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/RandomSeedGenerator.java (original)
+++ mahout/trunk/core/src/main/java/org/apache/mahout/clustering/kmeans/RandomSeedGenerator.java Thu Mar 31 09:25:25 2011
@@ -30,9 +30,11 @@ import org.apache.hadoop.io.SequenceFile
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.mahout.common.HadoopUtil;
+import org.apache.mahout.common.Pair;
import org.apache.mahout.common.RandomUtils;
import org.apache.mahout.common.distance.DistanceMeasure;
-import org.apache.mahout.math.Vector;
+import org.apache.mahout.common.iterator.sequencefile.PathFilters;
+import org.apache.mahout.common.iterator.sequencefile.SequenceFileIterable;
import org.apache.mahout.math.VectorWritable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,7 +43,6 @@ import org.slf4j.LoggerFactory;
* Given an Input Path containing a {@link org.apache.hadoop.io.SequenceFile}, randomly select k vectors and
* write them to the output file as a {@link org.apache.mahout.clustering.kmeans.Cluster} representing the
* initial centroid to use.
- * <p/>
*/
public final class RandomSeedGenerator {
@@ -52,13 +53,11 @@ public final class RandomSeedGenerator {
private RandomSeedGenerator() {
}
- public static Path buildRandom(Path input, Path output, int k, DistanceMeasure measure) throws IOException,
- IllegalAccessException,
- InstantiationException {
+ public static Path buildRandom(Path input, Path output, int k, DistanceMeasure measure) throws IOException {
// delete the output directory
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(output.toUri(), conf);
- HadoopUtil.overwriteOutput(output);
+ HadoopUtil.delete(conf, output);
Path outFile = new Path(output, "part-randomSeed");
boolean newFile = fs.createNewFile(outFile);
if (newFile) {
@@ -70,7 +69,7 @@ public final class RandomSeedGenerator {
inputPathPattern = input;
}
- FileStatus[] inputFiles = fs.globStatus(inputPathPattern);
+ FileStatus[] inputFiles = fs.globStatus(inputPathPattern, PathFilters.logsCRCFilter());
SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, outFile, Text.class, Cluster.class);
Random random = RandomUtils.getRandom();
List<Text> chosenTexts = new ArrayList<Text>(k);
@@ -78,13 +77,13 @@ public final class RandomSeedGenerator {
int nextClusterId = 0;
for (FileStatus fileStatus : inputFiles) {
- if (fileStatus.isDir() || fileStatus.getPath().getName().startsWith("_")) {
- continue; // select only the top level files that do not begin with "_" (Cloudera CHD3 adds _SUCCESS file)
+ if (fileStatus.isDir()) {
+ continue;
}
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, fileStatus.getPath(), conf);
- Writable key = reader.getKeyClass().asSubclass(Writable.class).newInstance();
- VectorWritable value = reader.getValueClass().asSubclass(VectorWritable.class).newInstance();
- while (reader.next(key, value)) {
+ for (Pair<Writable,VectorWritable> record :
+ new SequenceFileIterable<Writable,VectorWritable>(fileStatus.getPath(), true, conf)) {
+ Writable key = record.getFirst();
+ VectorWritable value = record.getSecond();
Cluster newCluster = new Cluster(value.get(), nextClusterId++, measure);
newCluster.observe(value.get(), 1);
Text newText = new Text(key.toString());
@@ -100,7 +99,6 @@ public final class RandomSeedGenerator {
chosenClusters.add(newCluster);
}
}
- reader.close();
}
for (int i = 0; i < k; i++) {
@@ -112,20 +110,5 @@ public final class RandomSeedGenerator {
return outFile;
}
-
- public static List<Vector> chooseRandomPoints(Iterable<Vector> vectors, int k) {
- List<Vector> chosenPoints = new ArrayList<Vector>(k);
- Random random = RandomUtils.getRandom();
- for (Vector value : vectors) {
- int currentSize = chosenPoints.size();
- if (currentSize < k) {
- chosenPoints.add(value);
- } else if (random.nextInt(currentSize + 1) == 0) { // with chance 1/(currentSize+1) pick new element
- int indexToRemove = random.nextInt(currentSize); // evict one chosen randomly
- chosenPoints.remove(indexToRemove);
- chosenPoints.add(value);
- }
- }
- return chosenPoints;
- }
+
}