You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by ro...@apache.org on 2010/02/13 21:27:30 UTC
svn commit: r909900 [4/4] - in
/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df: ./ builder/
callback/ data/ data/conditions/ mapred/ mapred/inmem/ mapred/partial/
mapreduce/ mapreduce/inmem/ mapreduce/partial/ node/ ref/ split/ tools/
Modified: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java?rev=909900&r1=909899&r2=909900&view=diff
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java (original)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java Sat Feb 13 20:27:25 2010
@@ -17,15 +17,21 @@
package org.apache.mahout.df.tools;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Arrays;
+
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.LongWritable;
-import org.apache.hadoop.io.SequenceFile.Reader;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.SequenceFile.Reader;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
@@ -42,38 +48,34 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.net.URI;
-import java.util.Arrays;
-
/**
* Temporary class used to compute the frequency distribution of the "class attribute".
*/
public class FrequenciesJob {
-
+
private static final Logger log = LoggerFactory.getLogger(FrequenciesJob.class);
-
+
/** directory that will hold this job's output */
private final Path outputPath;
-
+
/** file that contains the serialized dataset */
private final Path datasetPath;
-
+
/** directory that contains the data used in the first step */
private final Path dataPath;
-
+
/**
- * @param base base directory
- * @param dataPath data used in the first step
+ * @param base
+ * base directory
+ * @param dataPath
+ * data used in the first step
*/
public FrequenciesJob(Path base, Path dataPath, Path datasetPath) {
this.outputPath = new Path(base, "frequencies.output");
this.dataPath = dataPath;
this.datasetPath = datasetPath;
}
-
+
/**
* @return counts[partition][label] = num tuples from 'partition' with class == label
* @throws IOException
@@ -81,71 +83,71 @@
* @throws InterruptedException
*/
public int[][] run(Configuration conf) throws IOException, ClassNotFoundException, InterruptedException {
-
+
// check the output
FileSystem fs = outputPath.getFileSystem(conf);
- if (fs.exists(outputPath))
+ if (fs.exists(outputPath)) {
throw new IOException("Output path already exists : " + outputPath);
-
+ }
+
// put the dataset into the DistributedCache
- URI[] files = { datasetPath.toUri() };
+ URI[] files = {datasetPath.toUri()};
DistributedCache.setCacheFiles(files, conf);
-
+
Job job = new Job(conf);
job.setJarByClass(FrequenciesJob.class);
-
+
FileInputFormat.setInputPaths(job, dataPath);
FileOutputFormat.setOutputPath(job, outputPath);
-
+
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Frequencies.class);
-
+
job.setMapperClass(FrequenciesMapper.class);
job.setReducerClass(FrequenciesReducer.class);
-
+
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(SequenceFileOutputFormat.class);
-
+
// run the job
job.waitForCompletion(true);
-
+
int[][] counts = parseOutput(job);
-
-
+
// delete the output path
fs.delete(outputPath, true);
-
+
return counts;
}
-
+
/**
* Extracts the output and processes it
- *
+ *
* @return counts[partition][label] = num tuples from 'partition' with class == label
* @throws java.io.IOException
*/
protected int[][] parseOutput(JobContext job) throws IOException {
Configuration conf = job.getConfiguration();
-
+
int numMaps = conf.getInt("mapred.map.tasks", -1);
- log.info("mapred.map.tasks = {}", numMaps);
-
+ FrequenciesJob.log.info("mapred.map.tasks = {}", numMaps);
+
FileSystem fs = outputPath.getFileSystem(conf);
-
+
Path[] outfiles = DFUtils.listOutputFiles(fs, outputPath);
-
+
Frequencies[] values = new Frequencies[numMaps];
-
+
// read all the outputs
LongWritable key = new LongWritable();
Frequencies value = new Frequencies();
-
+
int index = 0;
for (Path path : outfiles) {
Reader reader = new Reader(fs, path, conf);
-
+
try {
while (reader.next(key, value)) {
values[index++] = value.clone();
@@ -154,35 +156,35 @@
reader.close();
}
}
-
+
if (index < numMaps) {
- throw new IllegalStateException(String.format("number of output Frequencies (%d) is lesser than the number of mappers!", index));
+ throw new IllegalStateException(String.format(
+ "number of output Frequencies (%d) is lesser than the number of mappers!", index));
}
-
+
// sort the frequencies using the firstIds
Arrays.sort(values);
return Frequencies.extractCounts(values);
}
-
+
/**
* Outputs the first key and the label of each tuple
- *
+ *
*/
- private static class FrequenciesMapper extends
- Mapper<LongWritable, Text, LongWritable, IntWritable> {
-
+ private static class FrequenciesMapper extends Mapper<LongWritable,Text,LongWritable,IntWritable> {
+
private LongWritable firstId;
-
+
private DataConverter converter;
-
+
@Override
protected void setup(Context context) throws IOException, InterruptedException {
Configuration conf = context.getConfiguration();
-
+
Dataset dataset = Builder.loadDataset(conf);
setup(dataset);
}
-
+
/**
* Useful when testing
*/
@@ -190,93 +192,93 @@
converter = new DataConverter(dataset);
// log.info("labels: {}", Arrays.toString(dataset.labels()));
}
-
+
@Override
- protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+ protected void map(LongWritable key, Text value, Context context) throws IOException,
+ InterruptedException {
if (firstId == null) {
firstId = new LongWritable(key.get());
}
-
- Instance instance = converter.convert((int)key.get(), value.toString());
-
+
+ Instance instance = converter.convert((int) key.get(), value.toString());
+
context.write(firstId, new IntWritable(instance.label));
}
-
+
}
-
- private static class FrequenciesReducer extends
- Reducer<LongWritable, IntWritable, LongWritable, Frequencies> {
-
+
+ private static class FrequenciesReducer extends Reducer<LongWritable,IntWritable,LongWritable,Frequencies> {
+
private int nblabels;
-
+
@Override
protected void setup(Context context) throws IOException, InterruptedException {
Configuration conf = context.getConfiguration();
- Dataset dataset = Builder.loadDataset(conf);
+ Dataset dataset = Builder.loadDataset(conf);
setup(dataset.nblabels());
}
-
+
/**
* Useful when testing
*/
protected void setup(int nblabels) {
this.nblabels = nblabels;
}
-
+
@Override
- protected void reduce(LongWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
+ protected void reduce(LongWritable key, Iterable<IntWritable> values, Context context) throws IOException,
+ InterruptedException {
int[] counts = new int[nblabels];
- for (IntWritable value : values){
+ for (IntWritable value : values) {
counts[value.get()]++;
}
-
+
context.write(key, new Frequencies(key.get(), counts));
}
}
+
/**
* Output of the job
- *
+ *
*/
- private static class Frequencies implements Writable,
- Comparable<Frequencies>, Cloneable {
-
+ private static class Frequencies implements Writable, Comparable<Frequencies>, Cloneable {
+
/**
* first key of the partition<br>
* used to sort the partitions
*/
private long firstId;
-
+
/** counts[c] = num tuples from the partition with label == c */
private int[] counts;
-
- protected Frequencies() {
- }
-
+
+ protected Frequencies() {}
+
protected Frequencies(long firstId, int[] counts) {
this.firstId = firstId;
this.counts = Arrays.copyOf(counts, counts.length);
}
-
+
protected long getFirstId() {
return firstId;
}
-
+
protected int[] getCounts() {
return counts;
}
-
+
@Override
public void readFields(DataInput in) throws IOException {
firstId = in.readLong();
counts = DFUtils.readIntArray(in);
}
-
+
@Override
public void write(DataOutput out) throws IOException {
out.writeLong(firstId);
DFUtils.writeArray(out, counts);
}
-
+
@Override
public boolean equals(Object other) {
if (!(other instanceof Frequencies)) {
@@ -284,44 +286,45 @@
}
return firstId == ((Frequencies) other).firstId;
}
-
+
@Override
public int hashCode() {
return (int) firstId;
}
-
+
@Override
protected Frequencies clone() {
return new Frequencies(firstId, counts);
}
-
+
@Override
public int compareTo(Frequencies obj) {
- if (firstId < obj.firstId)
+ if (firstId < obj.firstId) {
return -1;
- else if (firstId > obj.firstId)
+ } else if (firstId > obj.firstId) {
return 1;
- else
+ } else {
return 0;
+ }
}
-
+
public static int[] extractFirstIds(Frequencies[] partitions) {
int[] ids = new int[partitions.length];
-
+
for (int p = 0; p < partitions.length; p++) {
ids[p] = (int) partitions[p].firstId;
}
-
+
return ids;
}
-
+
public static int[][] extractCounts(Frequencies[] partitions) {
int[][] counts = new int[partitions.length][];
-
+
for (int p = 0; p < partitions.length; p++) {
counts[p] = partitions[p].counts;
}
-
+
return counts;
}
}
Modified: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java?rev=909900&r1=909899&r2=909900&view=diff
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java (original)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java Sat Feb 13 20:27:25 2010
@@ -1,5 +1,10 @@
package org.apache.mahout.df.tools;
+import java.io.File;
+import java.io.IOException;
+import java.util.Random;
+import java.util.Scanner;
+
import org.apache.commons.cli2.CommandLine;
import org.apache.commons.cli2.Group;
import org.apache.commons.cli2.Option;
@@ -22,131 +27,116 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.io.File;
-import java.util.Random;
-import java.util.Scanner;
-
/**
* This tool is used to uniformly distribute the class of all the tuples of the dataset over a given number of
* partitions.
*/
public class UDistrib {
-
+
private static final Logger log = LoggerFactory.getLogger(UDistrib.class);
-
- private UDistrib() {
- }
-
+
+ private UDistrib() {}
+
/**
* Launch the uniform distribution tool. Requires the following command line arguments:<br>
- *
- * data : data path
- * dataset : dataset path
- * numpartitions : num partitions
- * output : output path
- *
+ *
+ * data : data path dataset : dataset path numpartitions : num partitions output : output path
+ *
*/
public static void main(String[] args) throws IOException {
-
+
DefaultOptionBuilder obuilder = new DefaultOptionBuilder();
ArgumentBuilder abuilder = new ArgumentBuilder();
GroupBuilder gbuilder = new GroupBuilder();
-
- Option dataOpt = obuilder.withLongName("data").withShortName("d")
- .withRequired(true).withArgument(
- abuilder.withName("data").withMinimum(1).withMaximum(1).create())
- .withDescription("Data path").create();
-
- Option datasetOpt = obuilder.withLongName("dataset").withShortName(
- "ds").withRequired(true).withArgument(
- abuilder.withName("dataset").withMinimum(1).create())
- .withDescription("Dataset path").create();
-
- Option outputOpt = obuilder.withLongName("output").withShortName("o")
- .withRequired(true).withArgument(
- abuilder.withName("output").withMinimum(1).withMaximum(1).create())
- .withDescription("Path to generated files").create();
-
- Option partitionsOpt = obuilder.withLongName("numpartitions").withShortName("p")
- .withRequired(true).withArgument(
- abuilder.withName("numparts").withMinimum(1).withMinimum(1).create())
- .withDescription("Number of partitions to create").create();
- Option helpOpt = obuilder.withLongName("help").withDescription(
- "Print out help").withShortName("h").create();
-
- Group group = gbuilder.withName("Options").withOption(dataOpt).withOption(
- outputOpt).withOption(datasetOpt).withOption(partitionsOpt).withOption(helpOpt).create();
-
+
+ Option dataOpt = obuilder.withLongName("data").withShortName("d").withRequired(true).withArgument(
+ abuilder.withName("data").withMinimum(1).withMaximum(1).create()).withDescription("Data path").create();
+
+ Option datasetOpt = obuilder.withLongName("dataset").withShortName("ds").withRequired(true).withArgument(
+ abuilder.withName("dataset").withMinimum(1).create()).withDescription("Dataset path").create();
+
+ Option outputOpt = obuilder.withLongName("output").withShortName("o").withRequired(true).withArgument(
+ abuilder.withName("output").withMinimum(1).withMaximum(1).create()).withDescription(
+ "Path to generated files").create();
+
+ Option partitionsOpt = obuilder.withLongName("numpartitions").withShortName("p").withRequired(true)
+ .withArgument(abuilder.withName("numparts").withMinimum(1).withMinimum(1).create()).withDescription(
+ "Number of partitions to create").create();
+ Option helpOpt = obuilder.withLongName("help").withDescription("Print out help").withShortName("h")
+ .create();
+
+ Group group = gbuilder.withName("Options").withOption(dataOpt).withOption(outputOpt).withOption(
+ datasetOpt).withOption(partitionsOpt).withOption(helpOpt).create();
+
try {
Parser parser = new Parser();
parser.setGroup(group);
CommandLine cmdLine = parser.parse(args);
-
+
if (cmdLine.hasOption(helpOpt)) {
CommandLineUtil.printHelp(group);
return;
}
-
+
String data = cmdLine.getValue(dataOpt).toString();
String dataset = cmdLine.getValue(datasetOpt).toString();
int numPartitions = Integer.parseInt(cmdLine.getValue(partitionsOpt).toString());
String output = cmdLine.getValue(outputOpt).toString();
-
- runTool(data, dataset, output, numPartitions);
+
+ UDistrib.runTool(data, dataset, output, numPartitions);
} catch (OptionException e) {
- log.warn(e.toString(), e);
+ UDistrib.log.warn(e.toString(), e);
CommandLineUtil.printHelp(group);
}
-
+
}
-
+
private static void runTool(String dataStr, String datasetStr, String output, int numPartitions) throws IOException {
-
+
Configuration conf = new Configuration();
-
+
if (numPartitions <= 0) {
- throw new IllegalArgumentException("numPartitions <= 0");
+ throw new IllegalArgumentException("numPartitions <= 0");
}
-
+
// make sure the output file does not exist
Path outputPath = new Path(output);
FileSystem fs = outputPath.getFileSystem(conf);
-
+
if (fs.exists(outputPath)) {
- throw new IllegalArgumentException("Output path already exists");
+ throw new IllegalArgumentException("Output path already exists");
}
-
+
// create a new file corresponding to each partition
-// Path workingDir = fs.getWorkingDirectory();
-// FileSystem wfs = workingDir.getFileSystem(conf);
-// File parentFile = new File(workingDir.toString());
-// File tempFile = FileUtil.createLocalTempFile(parentFile, "Parts", true);
-// File tempFile = File.createTempFile("df.tools.UDistrib","");
-// tempFile.deleteOnExit();
+ // Path workingDir = fs.getWorkingDirectory();
+ // FileSystem wfs = workingDir.getFileSystem(conf);
+ // File parentFile = new File(workingDir.toString());
+ // File tempFile = FileUtil.createLocalTempFile(parentFile, "Parts", true);
+ // File tempFile = File.createTempFile("df.tools.UDistrib","");
+ // tempFile.deleteOnExit();
File tempFile = FileUtil.createLocalTempFile(new File(""), "df.tools.UDistrib", true);
Path partsPath = new Path(tempFile.toString());
FileSystem pfs = partsPath.getFileSystem(conf);
-
+
Path[] partPaths = new Path[numPartitions];
FSDataOutputStream[] files = new FSDataOutputStream[numPartitions];
for (int p = 0; p < numPartitions; p++) {
partPaths[p] = new Path(partsPath, String.format("part.%03d", p));
files[p] = pfs.create(partPaths[p]);
}
-
+
Path datasetPath = new Path(datasetStr);
Dataset dataset = Dataset.load(conf, datasetPath);
-
+
// currents[label] = next partition file where to place the tuple
int[] currents = new int[dataset.nblabels()];
-
+
// currents is initialized randomly in the range [0, numpartitions[
Random random = RandomUtils.getRandom();
for (int c = 0; c < currents.length; c++) {
currents[c] = random.nextInt(numPartitions);
}
-
+
// foreach tuple of the data
Path dataPath = new Path(dataStr);
FileSystem ifs = dataPath.getFileSystem(conf);
@@ -154,51 +144,52 @@
Scanner scanner = new Scanner(input);
DataConverter converter = new DataConverter(dataset);
int nbInstances = dataset.nbInstances();
-
+
int id = 0;
while (scanner.hasNextLine()) {
- if ((id % 1000)==0) {
- log.info(String.format("progress : %d / %d", id, nbInstances));
+ if (id % 1000 == 0) {
+ UDistrib.log.info(String.format("progress : %d / %d", id, nbInstances));
}
String line = scanner.nextLine();
- if (line.isEmpty())
+ if (line.isEmpty()) {
continue; // skip empty lines
-
+ }
+
// write the tuple in files[tuple.label]
Instance instance = converter.convert(id++, line);
int label = instance.label;
files[currents[label]].writeBytes(line);
files[currents[label]].writeChar('\n');
-
+
// update currents
currents[label]++;
- if (currents[label] == numPartitions)
+ if (currents[label] == numPartitions) {
currents[label] = 0;
+ }
}
-
+
// close all the files.
scanner.close();
- for (FSDataOutputStream file : files)
+ for (FSDataOutputStream file : files) {
file.close();
-
+ }
+
// merge all output files
FileUtil.copyMerge(pfs, partsPath, fs, outputPath, true, conf, null);
-/*
- FSDataOutputStream joined = fs.create(new Path(outputPath, "uniform.data"));
- for (int p = 0; p < numPartitions; p++) {
- log.info("Joining part : {}", p);
- FSDataInputStream partStream = fs.open(partPaths[p]);
-
- IOUtils.copyBytes(partStream, joined, conf, false);
-
- partStream.close();
- }
-
- joined.close();
-
- fs.delete(partsPath, true);
-*/
+ /*
+ * FSDataOutputStream joined = fs.create(new Path(outputPath, "uniform.data")); for (int p = 0; p <
+ * numPartitions; p++) { log.info("Joining part : {}", p); FSDataInputStream partStream =
+ * fs.open(partPaths[p]);
+ *
+ * IOUtils.copyBytes(partStream, joined, conf, false);
+ *
+ * partStream.close(); }
+ *
+ * joined.close();
+ *
+ * fs.delete(partsPath, true);
+ */
}
-
+
}