You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by ad...@apache.org on 2009/12/13 20:52:28 UTC
svn commit: r890108 - in
/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df:
data/DataConverter.java data/Dataset.java
mapreduce/partial/PartialBuilder.java tools/Frequencies.java
tools/FrequenciesJob.java tools/UDistrib.java
Author: adeneche
Date: Sun Dec 13 19:52:27 2009
New Revision: 890108
URL: http://svn.apache.org/viewvc?rev=890108&view=rev
Log:
MAHOUT-126 - work in progress
Added:
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/Frequencies.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java
Modified:
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/DataConverter.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/Dataset.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java
Modified: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/DataConverter.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/DataConverter.java?rev=890108&r1=890107&r2=890108&view=diff
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/DataConverter.java (original)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/DataConverter.java Sun Dec 13 19:52:27 2009
@@ -18,15 +18,20 @@
package org.apache.mahout.df.data;
import java.util.StringTokenizer;
+import java.util.Arrays;
import org.apache.commons.lang.ArrayUtils;
import org.apache.mahout.matrix.DenseVector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Converts String to Instance using a Dataset
*/
public class DataConverter {
+ private static final Logger log = LoggerFactory.getLogger(DataConverter.class);
+
private final Dataset dataset;
public DataConverter(Dataset dataset) {
@@ -48,7 +53,7 @@
int aId = 0;
int label = -1;
for (int attr = 0; attr < nball; attr++) {
- String token = tokenizer.nextToken();
+ String token = tokenizer.nextToken().trim();
if (ArrayUtils.contains(dataset.getIgnored(), attr)) {
continue; // IGNORED
@@ -56,6 +61,11 @@
if (attr == dataset.getLabelId()) {
label = dataset.labelCode(token);
+ if (label == -1) {
+ log.error(String.format("label token: %s\ndataset.labels: %s",
+ token, Arrays.toString(dataset.labels())));
+ throw new IllegalStateException("Label value ("+token+") not known");
+ }
} else if (dataset.isNumerical(aId)) {
vector.set(aId++, Double.parseDouble(token));
} else {
@@ -64,8 +74,10 @@
}
}
- if (label == -1)
+ if (label == -1) {
+ log.error(String.format("Label not found, instance id : %d, \nstring : %s", id, string));
throw new IllegalStateException("Label not found!");
+ }
return new Instance(id, vector, label);
}
Modified: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/Dataset.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/Dataset.java?rev=890108&r1=890107&r2=890108&view=diff
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/Dataset.java (original)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/data/Dataset.java Sun Dec 13 19:52:27 2009
@@ -77,7 +77,11 @@
/** number of instances in the dataset */
private int nbInstances;
-
+
+ public String[] labels() {
+ return Arrays.copyOf(labels, labels.length);
+ }
+
public int nblabels() {
return labels.length;
}
Modified: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java?rev=890108&r1=890107&r2=890108&view=diff
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java (original)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java Sun Dec 13 19:52:27 2009
@@ -142,14 +142,18 @@
* @param job
* @param outputPath directory that contains the output of the job
* @param firstIds partitions' first ids in hadoop's order
- * @param keys
+ * @param keys can be null
+ * @param trees can be null
* @param callback can be null
* @throws IOException
*/
protected static void processOutput(Job job, Path outputPath,
int[] firstIds, TreeID[] keys, Node[] trees, PredictionCallback callback)
throws IOException {
- if (keys.length != trees.length) {
+ if ((keys != null && trees == null)||(keys == null && trees != null)) {
+ throw new IllegalArgumentException("if keys is null, trees should also be null");
+ }
+ if (keys != null && keys.length != trees.length) {
throw new IllegalArgumentException("keys.length != trees.length");
}
@@ -187,8 +191,8 @@
}
// make sure we got all the keys/values
- if (index != keys.length) {
- throw new IllegalStateException();
+ if (keys != null && index != keys.length) {
+ throw new IllegalStateException("Some key/values are missing from the output");
}
}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/Frequencies.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/Frequencies.java?rev=890108&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/Frequencies.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/Frequencies.java Sun Dec 13 19:52:27 2009
@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.df.tools;
+
+import org.apache.commons.cli2.CommandLine;
+import org.apache.commons.cli2.Group;
+import org.apache.commons.cli2.Option;
+import org.apache.commons.cli2.OptionException;
+import org.apache.commons.cli2.builder.ArgumentBuilder;
+import org.apache.commons.cli2.builder.DefaultOptionBuilder;
+import org.apache.commons.cli2.builder.GroupBuilder;
+import org.apache.commons.cli2.commandline.Parser;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.mahout.common.CommandLineUtil;
+import org.apache.mahout.df.tools.FrequenciesJob;
+import org.apache.mahout.df.data.DataUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * Compute the frequency distribution of the "class label"
+ */
+public class Frequencies extends Configured implements Tool {
+
+ private static final Logger log = LoggerFactory.getLogger(Frequencies.class);
+
+ private Frequencies() {
+ }
+
+ @Override
+ public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
+
+ DefaultOptionBuilder obuilder = new DefaultOptionBuilder();
+ ArgumentBuilder abuilder = new ArgumentBuilder();
+ GroupBuilder gbuilder = new GroupBuilder();
+
+ Option dataOpt = obuilder.withLongName("data").withShortName("d")
+ .withRequired(true).withArgument(
+ abuilder.withName("path").withMinimum(1).withMaximum(1).create())
+ .withDescription("Data path").create();
+
+ Option datasetOpt = obuilder.withLongName("dataset").withShortName(
+ "ds").withRequired(true).withArgument(
+ abuilder.withName("path").withMinimum(1).create())
+ .withDescription("dataset path").create();
+
+ Option helpOpt = obuilder.withLongName("help").withDescription(
+ "Print out help").withShortName("h").create();
+
+ Group group = gbuilder.withName("Options").withOption(dataOpt).withOption(
+ datasetOpt).withOption(helpOpt).create();
+
+ try {
+ Parser parser = new Parser();
+ parser.setGroup(group);
+ CommandLine cmdLine = parser.parse(args);
+
+ if (cmdLine.hasOption(helpOpt)) {
+ CommandLineUtil.printHelp(group);
+ return 0;
+ }
+
+ String dataPath = cmdLine.getValue(dataOpt).toString();
+ String datasetPath = cmdLine.getValue(datasetOpt).toString();
+
+ log.debug("Data path : " + dataPath);
+ log.debug("Dataset path : " + datasetPath);
+
+ runTool(dataPath, datasetPath);
+ } catch (OptionException e) {
+ System.err.println("Exception : " + e);
+ CommandLineUtil.printHelp(group);
+ }
+
+ return 0;
+ }
+
+ private void runTool(String data, String dataset)
+ throws IOException, ClassNotFoundException, InterruptedException {
+
+ FileSystem fs = FileSystem.get(getConf());
+ Path workingDir = fs.getWorkingDirectory();
+
+ Path dataPath = new Path(data);
+ Path datasetPath = new Path(dataset);
+
+ log.info("Computing the frequencies...");
+ FrequenciesJob job = new FrequenciesJob(new Path(workingDir, "output"), dataPath, datasetPath);
+
+ int[][] counts = job.run(getConf());
+
+ // compute the partitions' sizes
+ int numPartitions = counts.length;
+ int[] sizes = new int[numPartitions];
+ for (int p = 0; p < numPartitions; p++) {
+ sizes[p] = DataUtils.sum(counts[p]);
+ }
+
+ // outputing the frequencies
+ log.info("counts[partition][class]");
+ for (int p = 0; p < numPartitions; p++) {
+ log.info(Arrays.toString(counts[p]));
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new Configuration(), new Frequencies(), args);
+ }
+
+}
\ No newline at end of file
Added: 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=890108&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/FrequenciesJob.java Sun Dec 13 19:52:27 2009
@@ -0,0 +1,328 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.mahout.df.tools;
+
+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.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+import org.apache.mahout.df.DFUtils;
+import org.apache.mahout.df.data.DataConverter;
+import org.apache.mahout.df.data.Dataset;
+import org.apache.mahout.df.data.Instance;
+import org.apache.mahout.df.mapreduce.Builder;
+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
+ */
+ 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
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
+ */
+ public int[][] run(Configuration conf) throws IOException, ClassNotFoundException, InterruptedException {
+
+ // check the output
+ FileSystem fs = outputPath.getFileSystem(conf);
+ if (fs.exists(outputPath))
+ throw new IOException("Output path already exists : " + outputPath);
+
+ // put the dataset into the DistributedCache
+ 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(Job job) throws IOException {
+ Configuration conf = job.getConfiguration();
+
+ int numMaps = conf.getInt("mapred.map.tasks", -1);
+ 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();
+ }
+ } finally {
+ reader.close();
+ }
+ }
+
+ if (index < numMaps) {
+ 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
+ *
+ */
+ protected 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
+ */
+ protected void setup(Dataset dataset) {
+ converter = new DataConverter(dataset);
+ // log.info("labels: " + Arrays.toString(dataset.labels()));
+ }
+
+ @Override
+ 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());
+
+ context.write(firstId, new IntWritable(instance.label));
+ }
+
+ }
+
+ protected 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);
+ setup(dataset.nblabels());
+ }
+
+ /**
+ * Useful when testing
+ */
+ protected void setup(int nblabels) {
+ this.nblabels = nblabels;
+ }
+
+ @Override
+ protected void reduce(LongWritable key, Iterable<IntWritable> values, Reducer<LongWritable,
+ IntWritable, LongWritable, Frequencies>.Context context) throws IOException, InterruptedException {
+ int[] counts = new int[nblabels];
+ for (IntWritable value : values){
+ counts[value.get()]++;
+ }
+
+ context.write(key, new Frequencies(key.get(), counts));
+ }
+ }
+ /**
+ * Output of the job
+ *
+ */
+ protected static class Frequencies implements Writable,
+ Comparable<Frequencies> {
+
+ /**
+ * 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(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)) {
+ return false;
+ }
+ 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)
+ return -1;
+ else if (firstId > obj.firstId)
+ return 1;
+ 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;
+ }
+ }
+}
\ No newline at end of file
Added: 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=890108&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/tools/UDistrib.java Sun Dec 13 19:52:27 2009
@@ -0,0 +1,160 @@
+package org.apache.mahout.df.tools;
+
+import org.apache.commons.cli2.CommandLine;
+import org.apache.commons.cli2.Group;
+import org.apache.commons.cli2.Option;
+import org.apache.commons.cli2.OptionException;
+import org.apache.commons.cli2.builder.ArgumentBuilder;
+import org.apache.commons.cli2.builder.DefaultOptionBuilder;
+import org.apache.commons.cli2.builder.GroupBuilder;
+import org.apache.commons.cli2.commandline.Parser;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.mahout.common.CommandLineUtil;
+import org.apache.mahout.common.RandomUtils;
+import org.apache.mahout.df.data.DataConverter;
+import org.apache.mahout.df.data.Dataset;
+import org.apache.mahout.df.data.Instance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.Scanner;
+
+/**
+ * This tool is used to uniformely 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);
+
+ /**
+ * Launch the uniform distribution tool. Requires the following command line arguments:<br>
+ *
+ * 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();
+
+ 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);
+ } catch (OptionException e) {
+ System.err.println("Exception : " + e);
+ CommandLineUtil.printHelp(group);
+ }
+
+ }
+
+ private static void runTool(String dataStr, String datasetStr, String output, int numPartitions) throws IOException {
+
+ Configuration conf = new Configuration();
+
+ // TODO exception if numPArtitions <= 0
+
+ // create a new file corresponding to each partition
+ Path outputPath = new Path(output);
+ FileSystem fs = outputPath.getFileSystem(conf);
+ FSDataOutputStream[] files = new FSDataOutputStream[numPartitions];
+ for (int p = 0; p < numPartitions; p++) {
+ files[p] = fs.create(new Path(outputPath, String.format("part-%03d.data", 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);
+ FSDataInputStream input = ifs.open(dataPath);
+ Scanner scanner = new Scanner(input);
+ DataConverter converter = new DataConverter(dataset);
+
+ int id = 0;
+ while (scanner.hasNextLine()) {
+ if ((id % 1000)==0) {
+ log.info("currentId : " + id);
+ }
+
+ String line = scanner.nextLine();
+ 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)
+ currents[label] = 0;
+ }
+
+ // close all the files.
+ scanner.close();
+ for (FSDataOutputStream file : files)
+ file.close();
+ }
+
+}