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/09/30 07:29:24 UTC
svn commit: r820181 [1/3] - in /lucene/mahout/trunk:
core/src/main/java/org/apache/mahout/df/mapred/partial/
core/src/main/java/org/apache/mahout/df/mapreduce/partial/
core/src/test/java/org/apache/mahout/df/mapred/partial/
core/src/test/java/org/apach...
Author: adeneche
Date: Wed Sep 30 05:29:22 2009
New Revision: 820181
URL: http://svn.apache.org/viewvc?rev=820181&view=rev
Log:
MAHOUT-145 PartialData mapreduce Random Forests
Added:
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/PartialBuilder.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step0Job.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step1Mapper.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Job.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Mapper.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/package.html
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/InterResults.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step0Job.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step1Mapper.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Job.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Mapper.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/TreeID.java
lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/package.html
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/PartialBuilderTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/PartialOutputCollector.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/PartialSequentialBuilder.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/PartitionBugTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/Step0JobTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/Step1MapperTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapred/partial/Step2MapperTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/InterResultsTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/MockContext.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/PartialBuilderTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/PartialSequentialBuilder.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/PartitionBugTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/Step0JobTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/Step1MapperTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/Step2MapperTest.java
lucene/mahout/trunk/core/src/test/java/org/apache/mahout/df/mapreduce/partial/TreeIDTest.java
lucene/mahout/trunk/examples/src/main/java/org/apache/mahout/df/mapred/
lucene/mahout/trunk/examples/src/main/java/org/apache/mahout/df/mapred/BuildForest.java
lucene/mahout/trunk/examples/src/main/java/org/apache/mahout/df/mapreduce/
lucene/mahout/trunk/examples/src/main/java/org/apache/mahout/df/mapreduce/BuildForest.java
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/PartialBuilder.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/PartialBuilder.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/PartialBuilder.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/PartialBuilder.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,212 @@
+/**
+ * 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.mapred.partial;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile.Reader;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.mahout.df.DFUtils;
+import org.apache.mahout.df.DecisionForest;
+import org.apache.mahout.df.builder.TreeBuilder;
+import org.apache.mahout.df.callback.PredictionCallback;
+import org.apache.mahout.df.mapred.Builder;
+import org.apache.mahout.df.mapred.partial.Step0Job.Step0Output;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.mapreduce.partial.TreeID;
+import org.apache.mahout.df.node.Node;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Builds a random forest using partial data. Each mapper uses only the data
+ * given by its InputSplit
+ */
+public class PartialBuilder extends Builder {
+
+ private final static Logger log = LoggerFactory.getLogger(PartialBuilder.class);
+
+ /**
+ * Indicates if we should run the second step of the builder.<br>
+ * This parameter is only meant for debuging, so we keep it protected.
+ *
+ * @param conf
+ * @return
+ */
+ protected static boolean isStep2(Configuration conf) {
+ return conf.getBoolean("debug.mahout.rf.partial.step2", true);
+ }
+
+ /**
+ * Should run the second step of the builder ?
+ *
+ * @param conf
+ * @param value true to indicate that the second step will be launched
+ *
+ */
+ protected static void setStep2(Configuration conf, boolean value) {
+ conf.setBoolean("debug.mahout.rf.partial.step2", value);
+ }
+
+ public PartialBuilder(TreeBuilder treeBuilder, Path dataPath,
+ Path datasetPath, Long seed) {
+ this(treeBuilder, dataPath, datasetPath, seed, new Configuration());
+ }
+
+ public PartialBuilder(TreeBuilder treeBuilder, Path dataPath,
+ Path datasetPath, Long seed, Configuration conf) {
+ super(treeBuilder, dataPath, datasetPath, seed, conf);
+ }
+
+ @Override
+ protected void configureJob(JobConf job, int nbTrees, boolean oobEstimate)
+ throws IOException {
+ FileInputFormat.setInputPaths(job, dataPath);
+ FileOutputFormat.setOutputPath(job, getOutputPath(job));
+
+ job.setOutputKeyClass(TreeID.class);
+ job.setOutputValueClass(MapredOutput.class);
+
+ job.setMapperClass(Step1Mapper.class);
+ job.setNumReduceTasks(0); // no reducers
+
+ job.setInputFormat(TextInputFormat.class);
+ job.setOutputFormat(SequenceFileOutputFormat.class);
+
+ // if we are in 'local' mode, correct the number of maps
+ // or the mappers won't be able to compute the right indexes
+ String tracker = job.get("mapred.job.tracker", "local");
+ if ("local".equals(tracker)) {
+ log.warn("Hadoop running in 'local' mode, only one map task will be launched");
+ job.setNumMapTasks(1);
+ }
+ }
+
+ @Override
+ protected DecisionForest parseOutput(JobConf job, PredictionCallback callback)
+ throws IOException {
+ int numMaps = job.getNumMapTasks();
+ int numTrees = getNbTrees(job);
+
+ Path outputPath = getOutputPath(job);
+
+ log.info("Computing partitions' first ids...");
+ Step0Job step0 = new Step0Job(getOutputPath(job), dataPath, datasetPath);
+ Step0Output[] partitions = step0.run(conf);
+
+ log.info("Processing the output...");
+ TreeID[] keys = new TreeID[numTrees];
+ Node[] trees = new Node[numTrees];
+ int[] firstIds = Step0Output.extractFirstIds(partitions);
+ processOutput(job, outputPath, firstIds, keys, trees, callback);
+
+ // call the second step in order to complete the oob predictions
+ if (callback != null && numMaps > 1 && isStep2(conf)) {
+ log.info("*****************************");
+ log.info("Second Step");
+ log.info("*****************************");
+ Step2Job step2 = new Step2Job(getOutputPath(job), dataPath, datasetPath, partitions);
+
+ step2.run(job, keys, trees, callback);
+ }
+
+ return new DecisionForest(Arrays.asList(trees));
+ }
+
+ /**
+ * Processes the output from the output path.<br>
+ *
+ * @param job
+ * @param outputPath directory that contains the output of the job
+ * @param firstIds partitions' first ids in hadoop's order
+ * @param keys
+ * @param values
+ * @param callback can be null
+ * @throws IOException
+ */
+ protected static void processOutput(JobConf job, Path outputPath,
+ int[] firstIds, TreeID[] keys, Node[] trees, PredictionCallback callback)
+ throws IOException {
+ assert keys.length == trees.length : "keys.length != trees.length";
+
+ FileSystem fs = outputPath.getFileSystem(job);
+
+ Path[] outfiles = DFUtils.listOutputFiles(fs, outputPath);
+
+ // read all the outputs
+ TreeID key = new TreeID();
+ MapredOutput value = new MapredOutput();
+
+ int index = 0;
+ for (Path path : outfiles) {
+ Reader reader = new Reader(fs, path, job);
+
+ try {
+ while (reader.next(key, value)) {
+ if (keys != null) {
+ keys[index] = key.clone();
+ }
+
+ if (trees != null) {
+ trees[index] = value.getTree();
+ }
+
+ processOutput(firstIds, key, value, callback);
+
+ index++;
+ }
+ } finally {
+ reader.close();
+ }
+ }
+
+ // make sure we got all the keys/values
+ assert index == keys.length;
+ }
+
+ /**
+ * Process the output, extracting the trees and passing the predictions to the
+ * callback
+ *
+ * @param firstIds partitions' first ids in hadoop's order
+ * @param keys
+ * @param values
+ * @param callback
+ * @return
+ */
+ private static void processOutput(int[] firstIds, TreeID key,
+ MapredOutput value, PredictionCallback callback) {
+
+ if (callback != null) {
+ int[] predictions = value.getPredictions();
+
+ for (int instanceId = 0; instanceId < predictions.length; instanceId++) {
+ callback.prediction(key.treeId(), firstIds[key.partition()] + instanceId,
+ predictions[instanceId]);
+ }
+ }
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step0Job.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step0Job.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step0Job.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step0Job.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,308 @@
+/**
+ * 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.mapred.partial;
+
+import static org.apache.mahout.df.DFUtils.listOutputFiles;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Arrays;
+
+import org.apache.commons.lang.ArrayUtils;
+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.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.SequenceFile.Reader;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.hadoop.mapred.TextInputFormat;
+
+/**
+ * preparation step of the partial mapreduce builder. Computes some stats that
+ * will be used by the builder.
+ */
+public class Step0Job {
+
+ /** directory that will hold this job's output */
+ protected final Path outputPath;
+
+ /** file that contains the serialized dataset */
+ protected final Path datasetPath;
+
+ /** directory that contains the data used in the first step */
+ protected final Path dataPath;
+
+ /**
+ * @param conf
+ * @param base base directory
+ * @param dataPath data used in the first step
+ * @param datasetPath
+ */
+ public Step0Job(Path base, Path dataPath, Path datasetPath) {
+ this.outputPath = new Path(base, "step0.output");
+ this.dataPath = dataPath;
+ this.datasetPath = datasetPath;
+ }
+
+ /**
+ * Computes the partitions' first ids in Hadoop's order
+ *
+ * @param conf configuration
+ * @return first ids for all the partitions
+ * @throws IOException
+ */
+ public Step0Output[] run(Configuration conf) throws IOException {
+
+ JobConf job = new JobConf(conf, Step0Job.class);
+
+ // check the output
+ if (outputPath.getFileSystem(job).exists(outputPath))
+ throw new RuntimeException("Ouput path already exists : " + outputPath);
+
+ // put the dataset into the DistributedCache
+ // use setCacheFiles() to overwrite the first-step cache files
+ URI[] files = new URI[] { datasetPath.toUri() };
+ DistributedCache.setCacheFiles(files, job);
+
+ FileInputFormat.setInputPaths(job, dataPath);
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ job.setOutputKeyClass(IntWritable.class);
+ job.setOutputValueClass(Step0Output.class);
+
+ job.setMapperClass(Step0Mapper.class);
+ job.setNumReduceTasks(0); // no reducers
+
+ job.setInputFormat(TextInputFormat.class);
+ job.setOutputFormat(SequenceFileOutputFormat.class);
+
+ // run the job
+ JobClient.runJob(job);
+
+ return parseOutput(job);
+ }
+
+ /**
+ * Extracts the output and processes it
+ *
+ * @param job
+ *
+ * @return firstIds for each partition in Hadoop's order
+ * @throws IOException
+ */
+ protected Step0Output[] parseOutput(JobConf job) throws IOException {
+ int numMaps = job.getNumMapTasks();
+ FileSystem fs = outputPath.getFileSystem(job);
+
+ Path[] outfiles = listOutputFiles(fs, outputPath);
+
+ int[] keys = new int[numMaps];
+ Step0Output[] values = new Step0Output[numMaps];
+
+ // read all the outputs
+ IntWritable key = new IntWritable();
+ Step0Output value = new Step0Output(0L, 0);
+
+ int index = 0;
+ for (Path path : outfiles) {
+ Reader reader = new Reader(fs, path, job);
+
+ try {
+ while (reader.next(key, value)) {
+ keys[index] = key.get();
+ values[index] = value.clone();
+
+ index++;
+ }
+ } finally {
+ reader.close();
+ }
+ }
+
+ return processOutput(keys, values);
+ }
+
+ /**
+ * Replaces the first id for each partition in Hadoop's order
+ *
+ * @param keys
+ * @param values
+ * @return
+ */
+ protected static Step0Output[] processOutput(int[] keys, Step0Output[] values) {
+ int numMaps = values.length;
+
+ // sort the values using firstId
+ Step0Output[] sorted = Arrays.copyOf(values, numMaps);
+ Arrays.sort(sorted);
+
+ // compute the partitions firstIds (file order)
+ int[] orderedIds = new int[numMaps];
+ orderedIds[0] = 0;
+ for (int p = 1; p < numMaps; p++) {
+ orderedIds[p] = orderedIds[p - 1] + sorted[p - 1].size;
+ }
+
+ // update the values' first ids
+ for (int p = 0; p < numMaps; p++) {
+ int order = ArrayUtils.indexOf(sorted, values[p]);
+ values[p].firstId = orderedIds[order];
+ }
+
+ // reorder the values in hadoop's order
+ Step0Output[] reordered = new Step0Output[numMaps];
+ for (int p = 0; p < numMaps; p++) {
+ reordered[keys[p]] = values[p];
+ }
+
+ return reordered;
+ }
+
+ /**
+ * Outputs the first key and the size of the partition
+ *
+ */
+ protected static class Step0Mapper extends MapReduceBase implements
+ Mapper<LongWritable, Text, IntWritable, Step0Output> {
+
+ protected int partition;
+
+ protected int size;
+
+ protected Long firstId;
+
+ protected OutputCollector<IntWritable, Step0Output> collector;
+
+ @Override
+ public void configure(JobConf job) {
+ configure(job.getInt("mapred.task.partition", -1));
+ }
+
+ /**
+ * Useful when testing
+ *
+ * @param p
+ */
+ protected void configure(int p) {
+ partition = p;
+ if (partition < 0) {
+ throw new RuntimeException("Wrong partition id : " + partition);
+ }
+ }
+
+ public void map(LongWritable key, Text value,
+ OutputCollector<IntWritable, Step0Output> output, Reporter reporter)
+ throws IOException {
+ if (firstId == null) {
+ firstId = key.get();
+ }
+
+ if (collector == null) {
+ collector = output;
+ }
+
+ size++;
+ }
+
+ @Override
+ public void close() throws IOException {
+ collector.collect(new IntWritable(partition), new Step0Output(firstId,
+ size));
+ }
+
+ }
+
+ /**
+ * Output of the step0's mappers
+ *
+ */
+ protected static class Step0Output implements Writable,
+ Comparable<Step0Output> {
+
+ /**
+ * first key of the partition<br>
+ * used to sort the partition
+ */
+ protected long firstId;
+
+ /** number of instances in the partition */
+ protected int size;
+
+ public Step0Output(long firstId, int size) {
+ this.firstId = firstId;
+ this.size = size;
+ }
+
+ public void readFields(DataInput in) throws IOException {
+ firstId = in.readLong();
+ size = in.readInt();
+ }
+
+ public void write(DataOutput out) throws IOException {
+ out.writeLong(firstId);
+ out.writeInt(size);
+ }
+
+ protected Step0Output clone() {
+ return new Step0Output(firstId, size);
+ }
+
+ public int compareTo(Step0Output obj) {
+ if (firstId < obj.firstId)
+ return -1;
+ else if (firstId > obj.firstId)
+ return 1;
+ else
+ return 0;
+ }
+
+ public static int[] extractFirstIds(Step0Output[] 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[] extractSizes(Step0Output[] partitions) {
+ int[] sizes = new int[partitions.length];
+
+ for (int p = 0; p < partitions.length; p++) {
+ sizes[p] = (int) partitions[p].size;
+ }
+
+ return sizes;
+ }
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step1Mapper.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step1Mapper.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step1Mapper.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step1Mapper.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,179 @@
+/**
+ * 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.mapred.partial;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.mahout.df.Bagging;
+import org.apache.mahout.df.callback.SingleTreePredictions;
+import org.apache.mahout.df.data.Data;
+import org.apache.mahout.df.data.DataConverter;
+import org.apache.mahout.df.data.Instance;
+import org.apache.mahout.df.mapred.Builder;
+import org.apache.mahout.df.mapred.MapredMapper;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.mapreduce.partial.TreeID;
+import org.apache.mahout.df.node.Node;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * First step of the Partial Data Builder. Builds the trees using the data
+ * available in the InputSplit. Predict the oob classes for each tree in its
+ * growing partition (input split).
+ */
+public class Step1Mapper extends MapredMapper implements
+ Mapper<LongWritable, Text, TreeID, MapredOutput> {
+
+ private static final Logger log = LoggerFactory.getLogger(Step1Mapper.class);
+
+ /** used to convert input values to data instances */
+ protected DataConverter converter;
+
+ protected Random rng;
+
+ /** number of trees to be built by this mapper */
+ protected int nbTrees;
+
+ /** id of the first tree */
+ protected int firstTreeId;
+
+ /** mapper's partition */
+ protected int partition;
+
+ /** used by close() */
+ protected OutputCollector<TreeID, MapredOutput> output;
+
+ /** will contain all instances if this mapper's split */
+ protected final List<Instance> instances = new ArrayList<Instance>();
+
+ @Override
+ public void configure(JobConf job) {
+ super.configure(job);
+
+ configure(Builder.getRandomSeed(job), job.getInt("mapred.task.partition",
+ -1), job.getNumMapTasks(), Builder.getNbTrees(job));
+ }
+
+ /**
+ * Useful when testing
+ *
+ * @param seed
+ * @param partition current mapper inputSplit partition
+ * @param numMapTasks number of running map tasks
+ * @param numTrees total number of trees in the forest
+ */
+ protected void configure(Long seed, int partition, int numMapTasks,
+ int numTrees) {
+ converter = new DataConverter(getDataset());
+
+ // prepare random-numders generator
+ log.debug("seed : " + seed);
+ if (seed == null)
+ rng = new Random();
+ else
+ rng = new Random(seed);
+
+ // mapper's partition
+ if (partition < 0) {
+ throw new RuntimeException("Wrong partition ID");
+ }
+ this.partition = partition;
+
+ // compute number of trees to build
+ nbTrees = nbTrees(numMapTasks, numTrees, partition);
+
+ // compute first tree id
+ firstTreeId = 0;
+ for (int p = 0; p < partition; p++) {
+ firstTreeId += nbTrees(numMapTasks, numTrees, p);
+ }
+
+ log.debug("partition : " + partition);
+ log.debug("nbTrees : " + nbTrees);
+ log.debug("firstTreeId : " + firstTreeId);
+ }
+
+ /**
+ * Compute the number of trees for a given partition. The first partition (0)
+ * may be longer than the rest of partition because of the remainder.
+ *
+ * @param numMaps total number of maps (partitions)
+ * @param numTrees total number of trees to build
+ * @param partition partition to compute the number of trees for
+ * @return
+ */
+ public static int nbTrees(int numMaps, int numTrees, int partition) {
+ int nbTrees = (int) Math.floor(numTrees / numMaps);
+ if (partition == 0) {
+ nbTrees += (int) numTrees - nbTrees * numMaps;
+ }
+
+ return nbTrees;
+ }
+
+ public void map(LongWritable key, Text value,
+ OutputCollector<TreeID, MapredOutput> output, Reporter reporter)
+ throws IOException {
+ if (this.output == null) {
+ this.output = output;
+ }
+
+ instances.add(converter.convert((int) key.get(), value.toString()));
+ }
+
+ @Override
+ public void close() throws IOException {
+ // prepare the data
+ log.debug("partition: " + partition + "numInstances: " + instances.size());
+
+ Data data = new Data(getDataset(), instances);
+ Bagging bagging = new Bagging(getTreeBuilder(), data);
+
+ SingleTreePredictions callback = null;
+ int[] predictions = null;
+ TreeID key = new TreeID();
+
+ log.debug("Building " + nbTrees + " trees");
+ for (int treeId = 0; treeId < nbTrees; treeId++) {
+ log.debug("Building tree N° : " + treeId);
+ if (isOobEstimate() && !isNoOutput()) {
+ callback = new SingleTreePredictions(data.size());
+ predictions = callback.predictions;
+ }
+
+ Node tree = bagging.build(treeId, rng, callback);
+
+ key.set(partition, firstTreeId + treeId);
+
+ if (!isNoOutput()) {
+ MapredOutput emOut = new MapredOutput(tree, predictions);
+ output.collect(key, emOut);
+ }
+ }
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Job.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Job.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Job.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Job.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,151 @@
+/**
+ * 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.mapred.partial;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.mahout.df.callback.PredictionCallback;
+import org.apache.mahout.df.mapred.Builder;
+import org.apache.mahout.df.mapred.partial.Step0Job.Step0Output;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.mapreduce.partial.InterResults;
+import org.apache.mahout.df.mapreduce.partial.TreeID;
+import org.apache.mahout.df.node.Node;
+
+/**
+ * 2nd step of the partial mapreduce builder. Computes the oob predictions using
+ * all the trees of the forest
+ */
+public class Step2Job {
+
+ /** directory that will hold this job's output */
+ protected final Path outputPath;
+
+ /** file that will contains the forest, passed to the maps */
+ protected final Path forestPath;
+
+ /** file that contains the serialized dataset */
+ protected final Path datasetPath;
+
+ /** directory that contains the data used in the first step */
+ protected final Path dataPath;
+
+ /** partitions info in Hadoop's order */
+ protected final Step0Output[] partitions;
+
+ /**
+ * @param conf
+ * @param base base directory
+ * @param dataPath data used in the first step
+ * @param datasetPath
+ * @param partitions partitions' infos in hadoop's order
+ */
+ public Step2Job(Path base, Path dataPath, Path datasetPath, Step0Output[] partitions) {
+ this.outputPath = new Path(base, "step2.output");
+ this.forestPath = new Path(base, "forest");
+ this.dataPath = dataPath;
+ this.datasetPath = datasetPath;
+ this.partitions = partitions;
+ }
+
+ /**
+ * Run the second step.
+ *
+ * @param conf configuration
+ * @param keys keys returned by the first step
+ * @param trees trees returned by the first step
+ * @param callback
+ * @throws IOException
+ */
+ public void run(Configuration conf, TreeID[] keys, Node[] trees, PredictionCallback callback)
+ throws IOException {
+ if (callback == null) {
+ // no need to launch the job
+ return;
+ }
+
+ int numTrees = keys.length;
+
+ JobConf job = new JobConf(conf, Step2Job.class);
+
+ // check the output
+ if (outputPath.getFileSystem(job).exists(outputPath))
+ throw new RuntimeException("Ouput path already exists : " + outputPath);
+
+ int[] sizes = Step0Output.extractSizes(partitions);
+
+ InterResults.store(forestPath.getFileSystem(job), forestPath, keys, trees, sizes);
+
+ // needed by the mapper
+ Builder.setNbTrees(job, numTrees);
+
+ // put the dataset and the forest into the DistributedCache
+ // use setCacheFiles() to overwrite the first-step cache files
+ URI[] files = new URI[] { datasetPath.toUri(), forestPath.toUri() };
+ DistributedCache.setCacheFiles(files, job);
+
+ FileInputFormat.setInputPaths(job, dataPath);
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ job.setOutputKeyClass(TreeID.class);
+ job.setOutputValueClass(MapredOutput.class);
+
+ job.setMapperClass(Step2Mapper.class);
+ job.setNumReduceTasks(0); // no reducers
+
+ job.setInputFormat(TextInputFormat.class);
+ job.setOutputFormat(SequenceFileOutputFormat.class);
+
+ // run the job
+ JobClient.runJob(job);
+
+ parseOutput(job, callback);
+ }
+
+ /**
+ * Extracts the output and processes it
+ *
+ * @param job
+ * @param callback
+ * @throws IOException
+ */
+ protected void parseOutput(JobConf job, PredictionCallback callback)
+ throws IOException {
+ int numMaps = job.getNumMapTasks();
+ int numTrees = Builder.getNbTrees(job);
+
+ // compute the total number of output values
+ int total = 0;
+ for (int partition = 0; partition < numMaps; partition++) {
+ total += Step2Mapper.nbConcerned(numMaps, numTrees, partition);
+ }
+
+ int[] firstIds = Step0Output.extractFirstIds(partitions);
+ PartialBuilder.processOutput(job, outputPath, firstIds, null, null, callback);
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Mapper.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Mapper.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Mapper.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/Step2Mapper.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,203 @@
+/**
+ * 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.mapred.partial;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.mahout.df.callback.SingleTreePredictions;
+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.mapred.Builder;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.mapreduce.partial.InterResults;
+import org.apache.mahout.df.mapreduce.partial.TreeID;
+import org.apache.mahout.df.node.Node;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Second step of PartialBuilder. Using the trees of the first step, computes
+ * the oob predictions for each tree, except those of its own partition, on all
+ * instancesof the partition.
+ */
+public class Step2Mapper extends MapReduceBase implements
+ Mapper<LongWritable, Text, TreeID, MapredOutput> {
+
+ private static final Logger log = LoggerFactory.getLogger(Step2Mapper.class);
+
+ protected TreeID[] keys;
+
+ protected Node[] trees;
+
+ protected SingleTreePredictions[] callbacks;
+
+ protected DataConverter converter;
+
+ protected int partition = -1;
+
+ /** used by close() */
+ protected OutputCollector<TreeID, MapredOutput> output;
+
+ /** num treated instances */
+ protected int instanceId;
+
+ @Override
+ public void configure(JobConf job) {
+ // get the cached files' paths
+ URI[] files = new URI[0];
+ try {
+ files = DistributedCache.getCacheFiles(job);
+ } catch (IOException e) {
+ throw new RuntimeException("Exception while getting the cache files : "
+ + e.getMessage());
+ }
+
+ if (files == null || files.length < 2) {
+ throw new RuntimeException("missing paths from the DistributedCache");
+ }
+
+ Dataset dataset;
+ try {
+ Path datasetPath = new Path(files[0].getPath());
+ dataset = Dataset.load(job, datasetPath);
+ } catch (IOException e) {
+ throw new RuntimeException("Exception while loading the dataset : "
+ + e.getMessage());
+ }
+
+ int numMaps = job.getNumMapTasks();
+ int p = job.getInt("mapred.task.partition", -1);
+
+ // total number of trees in the forest
+ int numTrees = Builder.getNbTrees(job);
+ if (numTrees == -1) {
+ throw new RuntimeException("numTrees not found !");
+ }
+
+ int nbConcerned = nbConcerned(numMaps, numTrees, p);
+ keys = new TreeID[nbConcerned];
+ trees = new Node[nbConcerned];
+
+ int numInstances;
+
+ try {
+ Path forestPath = new Path(files[1].getPath());
+ FileSystem fs = forestPath.getFileSystem(job);
+ numInstances = InterResults.load(fs, forestPath, numMaps, numTrees,
+ p, keys, trees);
+
+ log.debug("partition: " + p + "numInstances: " + numInstances);
+ } catch (IOException e) {
+ throw new RuntimeException("Exception while loading the forest : "
+ + e.getMessage());
+ }
+
+ configure(p, dataset, keys, trees, numInstances);
+ }
+
+ /**
+ * Compute the number of trees that need to classify the instances of this
+ * mapper's partition
+ *
+ * @param numMaps total number of map tasks
+ * @param numTrees total number of trees in the forest
+ * @param partition mapper's partition
+ * @return
+ */
+ public static int nbConcerned(int numMaps, int numTrees, int partition) {
+ assert partition >= 0 : "partition < 0";
+ // the trees of the mapper's partition are not concerned
+ return numTrees - Step1Mapper.nbTrees(numMaps, numTrees, partition);
+ }
+
+ /**
+ * Useful for testing. Configures the mapper without using a JobConf<br>
+ * TODO we don't need the keys partitions, the tree ids should suffice
+ *
+ * @param partition mapper's partition
+ * @param dataset
+ * @param keys keys returned by the first step
+ * @param trees trees returned by the first step
+ * @param numInstances number of instances in the mapper's partition
+ */
+ public void configure(int partition, Dataset dataset, TreeID[] keys,
+ Node[] trees, int numInstances) {
+ this.partition = partition;
+ if (partition < 0) {
+ throw new RuntimeException("Wrong partition id : " + partition);
+ }
+
+ converter = new DataConverter(dataset);
+
+ assert keys.length == trees.length : "keys.length != trees.length";
+ int nbConcerned = keys.length;
+
+ this.keys = keys;
+ this.trees = trees;
+
+ // make sure the trees are not from this partition
+ for (TreeID key : keys) {
+ assert key.partition() != partition : "a tree from this partition was found !";
+ }
+
+ // init the callbacks
+ callbacks = new SingleTreePredictions[nbConcerned];
+ for (int index = 0; index < nbConcerned; index++) {
+ callbacks[index] = new SingleTreePredictions(numInstances);
+ }
+
+ }
+
+ public void map(LongWritable key, Text value,
+ OutputCollector<TreeID, MapredOutput> output, Reporter reporter)
+ throws IOException {
+ if (this.output == null) {
+ this.output = output;
+ }
+
+ Instance instance = converter.convert(instanceId, value.toString());
+
+ for (int index = 0; index < keys.length; index++) {
+ int prediction = trees[index].classify(instance);
+ callbacks[index].prediction(index, instanceId, prediction);
+ }
+
+ instanceId++;
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (int index = 0; index < keys.length; index++) {
+ TreeID key = new TreeID(partition, keys[index].treeId());
+ output.collect(key, new MapredOutput(callbacks[index].predictions));
+ }
+ }
+
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/package.html
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/package.html?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/package.html (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapred/partial/package.html Wed Sep 30 05:29:22 2009
@@ -0,0 +1,41 @@
+<HTML>
+ <!--
+ /**
+ * 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.
+ */
+
+ --><HEAD>
+ <TITLE>org.apache.mahout.rf.mapred.partial</TITLE>
+</HEAD>
+<BODY>
+<DIV align="center">
+Partial-data mapreduce implementation of <b>Random Decision Forests</b>
+</DIV>
+<DIV> </DIV>
+<DIV>
+The builder splits the data, using a <b>FileInputSplit</b>, among the mappers. Building the forest and estimating the oob error takes two job steps.<br>
+<br>
+In the first step, each mapper is responsible for growing a number of trees with its partition's, loading the data instances in its map() function, then building the trees in the close() method. It uses the reference implementation's code to build each tree and estimate the oob error.<br>
+<br>
+The second step is needed when estimating the oob error. Each mapper loads all the trees that does not belong to its own partition (were not built using the partition's data) and uses them to classify the partition's data instances. The data instances are loaded in the map() method and the classification is performed in the close() method.<br>
+<br>
+</DIV>
+<DIV> </DIV>
+<DIV align="center">
+Copyright © 2009 <A HREF="http://lucene.apache.org/mahout">Apache Software Foundation - Mahout</A>
+</DIV>
+</BODY>
+</HTML>
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/InterResults.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/InterResults.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/InterResults.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/InterResults.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,123 @@
+/**
+ * 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.mapreduce.partial;
+
+import java.io.IOException;
+
+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.df.node.Node;
+
+/**
+ * Stores/Loads the intermediate results of step1 needed by step2.<br>
+ * This class should not be needed outside of the partial package, so all its
+ * methods are protected.<br>
+ */
+public class InterResults {
+
+ /**
+ * Load the trees and the keys returned from the first step
+ *
+ * @param fs forest path file system
+ * @param forestPath file path to the (key,tree) file
+ * @param numMaps number of map tasks
+ * @param numTrees total number of trees in the forest
+ * @param partition current partition
+ * @param keys array of size numTrees, will contain the loaded keys
+ * @param trees array of size numTrees, will contain the loaded trees
+ * @return number of instances in the current partition
+ * @throws IOException
+ */
+ public static int load(FileSystem fs, Path forestPath, int numMaps,
+ int numTrees, int partition, TreeID[] keys, Node[] trees)
+ throws IOException {
+ assert keys.length == trees.length : "keys.length != trees.length";
+
+ FSDataInputStream in = fs.open(forestPath);
+
+ TreeID key = new TreeID();
+ int numInstances = -1;
+
+ try {
+ // get current partition's size
+ for (int p = 0; p < numMaps; p++) {
+ if (p == partition) {
+ numInstances = in.readInt();
+ } else {
+ in.readInt();
+ }
+ }
+
+ // load (key, tree)
+ int current = 0;
+ for (int index = 0; index < numTrees; index++) {
+ key.readFields(in);
+
+ if (key.partition() != partition) {
+ keys[current] = key.clone();
+ trees[current] = Node.read(in);
+
+ current++;
+ } else {
+ // skip the trees of the current partition
+ Node.read(in);
+ }
+ }
+
+ assert current == keys.length : "loaded less keys/trees than expected";
+ } finally {
+ in.close();
+ }
+
+ return numInstances;
+ }
+
+ /**
+ * Write the forest trees into a file
+ *
+ * @param fs File System
+ * @param keys keys returned by the first step
+ * @param trees trees returned by the first step
+ * @param sizes partitions' sizes in hadoop order
+ * @throws IOException
+ */
+ public static void store(FileSystem fs, Path forestPath,
+ TreeID[] keys, Node[] trees, int[] sizes) throws IOException {
+ assert keys.length == trees.length : "keys.length != trees.length";
+
+ int numTrees = keys.length;
+ int numMaps = sizes.length;
+
+ FSDataOutputStream out = fs.create(forestPath);
+
+ // write partitions' sizes
+ for (int p = 0; p < numMaps; p++) {
+ out.writeInt(sizes[p]);
+ }
+
+ // write the data
+ for (int index = 0; index < numTrees; index++) {
+ keys[index].write(out);
+ trees[index].write(out);
+ }
+
+ out.close();
+ }
+
+}
Added: 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=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/PartialBuilder.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,214 @@
+/**
+ * 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.mapreduce.partial;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile.Reader;
+import org.apache.hadoop.mapreduce.Job;
+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.DecisionForest;
+import org.apache.mahout.df.builder.TreeBuilder;
+import org.apache.mahout.df.callback.PredictionCallback;
+import org.apache.mahout.df.mapreduce.Builder;
+import org.apache.mahout.df.mapreduce.partial.Step0Job.Step0Output;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.node.Node;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Builds a random forest using partial data. Each mapper uses only the data
+ * given by its InputSplit
+ */
+public class PartialBuilder extends Builder {
+
+ private final static Logger log = LoggerFactory.getLogger(PartialBuilder.class);
+
+ /**
+ * Indicates if we should run the second step of the builder.<br>
+ * This parameter is only meant for debuging, so we keep it protected.
+ *
+ * @param conf
+ * @return
+ */
+ protected static boolean isStep2(Configuration conf) {
+ return conf.getBoolean("debug.mahout.rf.partial.step2", true);
+ }
+
+ /**
+ * Should run the second step of the builder ?
+ *
+ * @param conf
+ * @param value true to indicate that the second step will be launched
+ *
+ */
+ protected static void setStep2(Configuration conf, boolean value) {
+ conf.setBoolean("debug.mahout.rf.partial.step2", value);
+ }
+
+ public PartialBuilder(TreeBuilder treeBuilder, Path dataPath,
+ Path datasetPath, Long seed) {
+ this(treeBuilder, dataPath, datasetPath, seed, new Configuration());
+ }
+
+ public PartialBuilder(TreeBuilder treeBuilder, Path dataPath,
+ Path datasetPath, Long seed, Configuration conf) {
+ super(treeBuilder, dataPath, datasetPath, seed, conf);
+ }
+
+
+ @Override
+ protected void configureJob(Job job, int nbTrees, boolean oobEstimate)
+ throws IOException {
+ Configuration conf = job.getConfiguration();
+
+ job.setJarByClass(PartialBuilder.class);
+
+ FileInputFormat.setInputPaths(job, dataPath);
+ FileOutputFormat.setOutputPath(job, getOutputPath(conf));
+
+ job.setOutputKeyClass(TreeID.class);
+ job.setOutputValueClass(MapredOutput.class);
+
+ job.setMapperClass(Step1Mapper.class);
+ job.setNumReduceTasks(0); // no reducers
+
+ job.setInputFormatClass(TextInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ }
+
+ @Override
+ protected DecisionForest parseOutput(Job job, PredictionCallback callback)
+ throws Exception {
+ Configuration conf = job.getConfiguration();
+
+ int numTrees = getNbTrees(conf);
+
+ Path outputPath = getOutputPath(conf);
+
+ log.info("Computing partitions' first ids...");
+ Step0Job step0 = new Step0Job(getOutputPath(conf), dataPath, datasetPath);
+ Step0Output[] partitions = step0.run(new Configuration(conf));
+
+ log.info("Processing the output...");
+ TreeID[] keys = new TreeID[numTrees];
+ Node[] trees = new Node[numTrees];
+ int[] firstIds = Step0Output.extractFirstIds(partitions);
+ processOutput(job, outputPath, firstIds, keys, trees, callback);
+
+ // JobClient should have updated numMaps to the correct number of maps
+ int numMaps = partitions.length;
+
+ // call the second step in order to complete the oob predictions
+ if (callback != null && numMaps > 1 && isStep2(conf)) {
+ log.info("*****************************");
+ log.info("Second Step");
+ log.info("*****************************");
+ Step2Job step2 = new Step2Job(getOutputPath(conf), dataPath, datasetPath, partitions);
+
+ step2.run(new Configuration(conf), keys, trees, callback);
+ }
+
+ return new DecisionForest(Arrays.asList(trees));
+ }
+
+ /**
+ * Processes the output from the output path.<br>
+ *
+ * @param job
+ * @param outputPath directory that contains the output of the job
+ * @param firstIds partitions' first ids in hadoop's order
+ * @param keys
+ * @param values
+ * @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 {
+ assert keys.length == trees.length : "keys.length != trees.length";
+
+ Configuration conf = job.getConfiguration();
+
+ FileSystem fs = outputPath.getFileSystem(conf);
+
+ Path[] outfiles = DFUtils.listOutputFiles(fs, outputPath);
+
+ // read all the outputs
+ TreeID key = new TreeID();
+ MapredOutput value = new MapredOutput();
+
+ int index = 0;
+ for (Path path : outfiles) {
+ Reader reader = new Reader(fs, path, conf);
+
+ try {
+ while (reader.next(key, value)) {
+ if (keys != null) {
+ keys[index] = key.clone();
+ }
+
+ if (trees != null) {
+ trees[index] = value.getTree();
+ }
+
+ processOutput(firstIds, key, value, callback);
+
+ index++;
+ }
+ } finally {
+ reader.close();
+ }
+ }
+
+ // make sure we got all the keys/values
+ assert index == keys.length;
+ }
+
+ /**
+ * Process the output, extracting the trees and passing the predictions to the
+ * callback
+ *
+ * @param firstIds partitions' first ids in hadoop's order
+ * @param keys
+ * @param values
+ * @param callback
+ * @return
+ */
+ private static void processOutput(int[] firstIds, TreeID key,
+ MapredOutput value, PredictionCallback callback) {
+
+ if (callback != null) {
+ int[] predictions = value.getPredictions();
+
+ for (int instanceId = 0; instanceId < predictions.length; instanceId++) {
+ callback.prediction(key.treeId(), firstIds[key.partition()] + instanceId,
+ predictions[instanceId]);
+ }
+ }
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step0Job.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step0Job.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step0Job.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step0Job.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,304 @@
+/**
+ * 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.mapreduce.partial;
+
+import static org.apache.mahout.df.DFUtils.listOutputFiles;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang.ArrayUtils;
+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.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.Mapper;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * preparation step of the partial mapreduce builder. Computes some stats that
+ * will be used by the builder.
+ */
+public class Step0Job {
+
+ private static final Logger log = LoggerFactory.getLogger(Step0Job.class);
+
+ /** directory that will hold this job's output */
+ protected final Path outputPath;
+
+ /** file that contains the serialized dataset */
+ protected final Path datasetPath;
+
+ /** directory that contains the data used in the first step */
+ protected final Path dataPath;
+
+ /**
+ * @param conf
+ * @param base base directory
+ * @param dataPath data used in the first step
+ * @param datasetPath
+ */
+ public Step0Job(Path base, Path dataPath, Path datasetPath) {
+ this.outputPath = new Path(base, "step0.output");
+ this.dataPath = dataPath;
+ this.datasetPath = datasetPath;
+ }
+
+ /**
+ * Computes the partitions' info in Hadoop's order
+ *
+ * @param conf configuration
+ * @return partitions' info in Hadoop's order
+ * @throws Exception
+ */
+ public Step0Output[] run(Configuration conf) throws Exception {
+
+ // check the output
+ if (outputPath.getFileSystem(conf).exists(outputPath))
+ throw new RuntimeException("Ouput path already exists : " + outputPath);
+
+ // put the dataset into the DistributedCache
+ // use setCacheFiles() to overwrite the first-step cache files
+ URI[] files = new URI[] { datasetPath.toUri() };
+ DistributedCache.setCacheFiles(files, conf);
+
+ Job job = new Job(conf);
+ job.setJarByClass(Step0Job.class);
+
+ FileInputFormat.setInputPaths(job, dataPath);
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ job.setOutputKeyClass(IntWritable.class);
+ job.setOutputValueClass(Step0Output.class);
+
+ job.setMapperClass(Step0Mapper.class);
+ job.setNumReduceTasks(0); // no reducers
+
+ job.setInputFormatClass(TextInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+
+ // run the job
+ job.waitForCompletion(true);
+
+ return parseOutput(job);
+ }
+
+ /**
+ * Extracts the output and processes it
+ *
+ * @param conf
+ *
+ * @return info for each partition in Hadoop's order
+ * @throws IOException
+ */
+ protected Step0Output[] parseOutput(Job job) throws IOException {
+ Configuration conf = job.getConfiguration();
+
+ log.info("mapred.map.tasks = " + conf.getInt("mapred.map.tasks", -1));
+
+ FileSystem fs = outputPath.getFileSystem(conf);
+
+ Path[] outfiles = listOutputFiles(fs, outputPath);
+
+ List<Integer> keys = new ArrayList<Integer>();
+ List<Step0Output> values = new ArrayList<Step0Output>();
+
+ // read all the outputs
+ IntWritable key = new IntWritable();
+ Step0Output value = new Step0Output(0L, 0);
+
+ for (Path path : outfiles) {
+ Reader reader = new Reader(fs, path, conf);
+
+ try {
+ while (reader.next(key, value)) {
+ keys.add(key.get());
+ values.add(value.clone());
+ }
+ } finally {
+ reader.close();
+ }
+ }
+
+ return processOutput(keys, values);
+ }
+
+ /**
+ * Replaces the first id for each partition in Hadoop's order
+ *
+ * @param keys
+ * @param values
+ * @return
+ */
+ protected static Step0Output[] processOutput(List<Integer> keys, List<Step0Output> values) {
+ int numMaps = values.size();
+
+ // sort the values using firstId
+ Step0Output[] sorted = new Step0Output[numMaps];
+ values.toArray(sorted);
+ Arrays.sort(sorted);
+
+ // compute the partitions firstIds (file order)
+ int[] orderedIds = new int[numMaps];
+ orderedIds[0] = 0;
+ for (int p = 1; p < numMaps; p++) {
+ orderedIds[p] = orderedIds[p - 1] + sorted[p - 1].size;
+ }
+
+ // update the values' first ids
+ for (int p = 0; p < numMaps; p++) {
+ int order = ArrayUtils.indexOf(sorted, values.get(p));
+ values.get(p).firstId = orderedIds[order];
+ }
+
+ // reorder the values in hadoop's order
+ Step0Output[] reordered = new Step0Output[numMaps];
+ for (int p = 0; p < numMaps; p++) {
+ reordered[keys.get(p)] = values.get(p);
+ }
+
+ return reordered;
+ }
+
+ /**
+ * Outputs the first key and the size of the partition
+ *
+ */
+ protected static class Step0Mapper extends
+ Mapper<LongWritable, Text, IntWritable, Step0Output> {
+
+ protected int partition;
+
+ protected int size;
+
+ protected Long firstId;
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+ configure(context.getConfiguration().getInt("mapred.task.partition", -1));
+ }
+
+ /**
+ * Useful when testing
+ *
+ * @param p
+ */
+ protected void configure(int p) {
+ partition = p;
+ if (partition < 0) {
+ throw new RuntimeException("Wrong partition id : " + partition);
+ }
+ }
+
+ @Override
+ protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+ if (firstId == null) {
+ firstId = key.get();
+ }
+
+ size++;
+ }
+
+ @Override
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ context.write(new IntWritable(partition), new Step0Output(firstId, size));
+ }
+
+ }
+
+ /**
+ * Output of the step0's mappers
+ *
+ */
+ protected static class Step0Output implements Writable,
+ Comparable<Step0Output> {
+
+ /**
+ * first key of the partition<br>
+ * used to sort the partition
+ */
+ protected long firstId;
+
+ /** number of instances in the partition */
+ protected int size;
+
+ public Step0Output(long firstId, int size) {
+ this.firstId = firstId;
+ this.size = size;
+ }
+
+ public void readFields(DataInput in) throws IOException {
+ firstId = in.readLong();
+ size = in.readInt();
+ }
+
+ public void write(DataOutput out) throws IOException {
+ out.writeLong(firstId);
+ out.writeInt(size);
+ }
+
+ protected Step0Output clone() {
+ return new Step0Output(firstId, size);
+ }
+
+ public int compareTo(Step0Output obj) {
+ if (firstId < obj.firstId)
+ return -1;
+ else if (firstId > obj.firstId)
+ return 1;
+ else
+ return 0;
+ }
+
+ public static int[] extractFirstIds(Step0Output[] 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[] extractSizes(Step0Output[] partitions) {
+ int[] sizes = new int[partitions.length];
+
+ for (int p = 0; p < partitions.length; p++) {
+ sizes[p] = (int) partitions[p].size;
+ }
+
+ return sizes;
+ }
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step1Mapper.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step1Mapper.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step1Mapper.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step1Mapper.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,172 @@
+/**
+ * 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.mapreduce.partial;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.mahout.df.Bagging;
+import org.apache.mahout.df.callback.SingleTreePredictions;
+import org.apache.mahout.df.data.Data;
+import org.apache.mahout.df.data.DataConverter;
+import org.apache.mahout.df.data.Instance;
+import org.apache.mahout.df.mapreduce.Builder;
+import org.apache.mahout.df.mapreduce.MapredMapper;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.node.Node;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * First step of the Partial Data Builder. Builds the trees using the data
+ * available in the InputSplit. Predict the oob classes for each tree in its
+ * growing partition (input split).
+ */
+public class Step1Mapper extends
+ MapredMapper<LongWritable, Text, TreeID, MapredOutput> {
+
+ private static final Logger log = LoggerFactory.getLogger(Step1Mapper.class);
+
+ /** used to convert input values to data instances */
+ protected DataConverter converter;
+
+ protected Random rng;
+
+ /** number of trees to be built by this mapper */
+ protected int nbTrees;
+
+ /** id of the first tree */
+ protected int firstTreeId;
+
+ /** mapper's partition */
+ protected int partition;
+
+ /** will contain all instances if this mapper's split */
+ protected final List<Instance> instances = new ArrayList<Instance>();
+
+ @Override
+ protected void setup(Context context) throws IOException,
+ InterruptedException {
+ super.setup(context);
+ Configuration conf = context.getConfiguration();
+
+ configure(Builder.getRandomSeed(conf), conf.getInt("mapred.task.partition",
+ -1), Builder.getNumMaps(conf), Builder.getNbTrees(conf));
+ }
+
+ /**
+ * Useful when testing
+ *
+ * @param seed
+ * @param partition current mapper inputSplit partition
+ * @param numMapTasks number of running map tasks
+ * @param numTrees total number of trees in the forest
+ */
+ protected void configure(Long seed, int partition, int numMapTasks,
+ int numTrees) {
+ converter = new DataConverter(getDataset());
+
+ // prepare random-numders generator
+ log.debug("seed : " + seed);
+ if (seed == null)
+ rng = new Random();
+ else
+ rng = new Random(seed);
+
+ // mapper's partition
+ if (partition < 0) {
+ throw new RuntimeException("Wrong partition ID");
+ }
+ this.partition = partition;
+
+ // compute number of trees to build
+ nbTrees = nbTrees(numMapTasks, numTrees, partition);
+
+ // compute first tree id
+ firstTreeId = 0;
+ for (int p = 0; p < partition; p++) {
+ firstTreeId += nbTrees(numMapTasks, numTrees, p);
+ }
+
+ log.debug("partition : " + partition);
+ log.debug("nbTrees : " + nbTrees);
+ log.debug("firstTreeId : " + firstTreeId);
+ }
+
+ /**
+ * Compute the number of trees for a given partition. The first partition (0)
+ * may be longer than the rest of partition because of the remainder.
+ *
+ * @param numMaps total number of maps (partitions)
+ * @param numTrees total number of trees to build
+ * @param partition partition to compute the number of trees for
+ * @return
+ */
+ public static int nbTrees(int numMaps, int numTrees, int partition) {
+ int nbTrees = (int) Math.floor(numTrees / numMaps);
+ if (partition == 0) {
+ nbTrees += (int) numTrees - nbTrees * numMaps;
+ }
+
+ return nbTrees;
+ }
+
+
+ @Override
+ protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+ instances.add(converter.convert((int) key.get(), value.toString()));
+ }
+
+
+ @Override
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ // prepare the data
+ log.debug("partition: " + partition + "numInstances: " + instances.size());
+
+ Data data = new Data(getDataset(), instances);
+ Bagging bagging = new Bagging(getTreeBuilder(), data);
+
+ SingleTreePredictions callback = null;
+ int[] predictions = null;
+ TreeID key = new TreeID();
+
+ log.debug("Building " + nbTrees + " trees");
+ for (int treeId = 0; treeId < nbTrees; treeId++) {
+ log.debug("Building tree N° : " + treeId);
+ if (isOobEstimate() && !isNoOutput()) {
+ callback = new SingleTreePredictions(data.size());
+ predictions = callback.predictions;
+ }
+
+ Node tree = bagging.build(treeId, rng, callback);
+
+ key.set(partition, firstTreeId + treeId);
+
+ if (!isNoOutput()) {
+ MapredOutput emOut = new MapredOutput(tree, predictions);
+ context.write(key, emOut);
+ }
+ }
+ }
+
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Job.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Job.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Job.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Job.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,151 @@
+/**
+ * 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.mapreduce.partial;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+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.callback.PredictionCallback;
+import org.apache.mahout.df.mapreduce.Builder;
+import org.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.mapreduce.partial.Step0Job.Step0Output;
+import org.apache.mahout.df.node.Node;
+
+/**
+ * 2nd step of the partial mapreduce builder. Computes the oob predictions using
+ * all the trees of the forest
+ */
+public class Step2Job {
+
+ /** directory that will hold this job's output */
+ protected final Path outputPath;
+
+ /** file that will contains the forest, passed to the maps */
+ protected final Path forestPath;
+
+ /** file that contains the serialized dataset */
+ protected final Path datasetPath;
+
+ /** directory that contains the data used in the first step */
+ protected final Path dataPath;
+
+ /** partitions info in Hadoop's order */
+ protected final Step0Output[] partitions;
+
+ /**
+ * @param conf
+ * @param base base directory
+ * @param dataPath data used in the first step
+ * @param datasetPath
+ * @param partitions partitions' infos in hadoop's order
+ */
+ public Step2Job(Path base, Path dataPath, Path datasetPath, Step0Output[] partitions) {
+ this.outputPath = new Path(base, "step2.output");
+ this.forestPath = new Path(base, "forest");
+ this.dataPath = dataPath;
+ this.datasetPath = datasetPath;
+ this.partitions = partitions;
+ }
+
+ /**
+ * Run the second step.
+ *
+ * @param conf configuration
+ * @param keys keys returned by the first step
+ * @param trees trees returned by the first step
+ * @param callback
+ * @throws Exception
+ */
+ public void run(Configuration conf, TreeID[] keys, Node[] trees, PredictionCallback callback)
+ throws Exception {
+ if (callback == null) {
+ // no need to launch the job
+ return;
+ }
+
+ int numTrees = keys.length;
+
+ // check the output
+ if (outputPath.getFileSystem(conf).exists(outputPath))
+ throw new RuntimeException("Ouput path already exists : " + outputPath);
+
+ int[] sizes = Step0Output.extractSizes(partitions);
+
+ InterResults.store(forestPath.getFileSystem(conf), forestPath, keys, trees, sizes);
+
+ // needed by the mapper
+ Builder.setNbTrees(conf, numTrees);
+
+ // put the dataset and the forest into the DistributedCache
+ // use setCacheFiles() to overwrite the first-step cache files
+ URI[] files = new URI[] { datasetPath.toUri(), forestPath.toUri() };
+ DistributedCache.setCacheFiles(files, conf);
+
+ Job job = new Job(conf);
+ job.setJarByClass(Step2Job.class);
+
+ FileInputFormat.setInputPaths(job, dataPath);
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ job.setOutputKeyClass(TreeID.class);
+ job.setOutputValueClass(MapredOutput.class);
+
+ job.setMapperClass(Step2Mapper.class);
+ job.setNumReduceTasks(0); // no reducers
+
+ job.setInputFormatClass(TextInputFormat.class);
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+
+ // run the job
+ if (job.waitForCompletion(true)) {
+ parseOutput(job, callback);
+ }
+ }
+
+ /**
+ * Extracts the output and processes it
+ *
+ * @param job
+ * @param callback
+ * @throws IOException
+ */
+ protected void parseOutput(Job job, PredictionCallback callback)
+ throws IOException {
+ Configuration conf = job.getConfiguration();
+
+ int numMaps = Builder.getNumMaps(conf);
+ int numTrees = Builder.getNbTrees(conf);
+
+ // compute the total number of output values
+ int total = 0;
+ for (int partition = 0; partition < numMaps; partition++) {
+ total += Step2Mapper.nbConcerned(numMaps, numTrees, partition);
+ }
+
+ int[] firstIds = Step0Output.extractFirstIds(partitions);
+ PartialBuilder.processOutput(job, outputPath, firstIds, null, null, callback);
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Mapper.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Mapper.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Mapper.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/Step2Mapper.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,198 @@
+/**
+ * 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.mapreduce.partial;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.commons.lang.ArrayUtils;
+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.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.mahout.df.callback.SingleTreePredictions;
+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.apache.mahout.df.mapreduce.MapredOutput;
+import org.apache.mahout.df.node.Node;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Second step of PartialBuilder. Using the trees of the first step, computes
+ * the oob predictions for each tree, except those of its own partition, on all
+ * instancesof the partition.
+ */
+public class Step2Mapper extends Mapper<LongWritable, Text, TreeID, MapredOutput> {
+
+ private static final Logger log = LoggerFactory.getLogger(Step2Mapper.class);
+
+ protected TreeID[] keys;
+
+ protected Node[] trees;
+
+ protected SingleTreePredictions[] callbacks;
+
+ protected DataConverter converter;
+
+ protected int partition = -1;
+
+ /** num treated instances */
+ protected int instanceId;
+
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+ Configuration conf = context.getConfiguration();
+
+ // get the cached files' paths
+ URI[] files = new URI[0];
+ try {
+ files = DistributedCache.getCacheFiles(conf);
+ } catch (IOException e) {
+ throw new RuntimeException("Exception while getting the cache files : "
+ + e.getMessage());
+ }
+
+ log.info("DistributedCache.getCacheFiles(): " + ArrayUtils.toString(files));
+
+ if (files == null || files.length < 2) {
+ throw new RuntimeException("missing paths from the DistributedCache");
+ }
+
+ Dataset dataset;
+ try {
+ Path datasetPath = new Path(files[0].getPath());
+ dataset = Dataset.load(conf, datasetPath);
+ } catch (IOException e) {
+ throw new RuntimeException("Exception while loading the dataset : "
+ + e.getMessage());
+ }
+
+ int numMaps = Builder.getNumMaps(conf);
+ int p = conf.getInt("mapred.task.partition", -1);
+
+ // total number of trees in the forest
+ int numTrees = Builder.getNbTrees(conf);
+ if (numTrees == -1) {
+ throw new RuntimeException("numTrees not found !");
+ }
+
+ int nbConcerned = nbConcerned(numMaps, numTrees, p);
+ keys = new TreeID[nbConcerned];
+ trees = new Node[nbConcerned];
+
+ int numInstances;
+
+ try {
+ Path forestPath = new Path(files[1].getPath());
+ FileSystem fs = forestPath.getFileSystem(conf);
+ numInstances = InterResults.load(fs, forestPath, numMaps, numTrees,
+ p, keys, trees);
+
+ log.debug("partition: " + p + "numInstances: " + numInstances);
+ } catch (IOException e) {
+ throw new RuntimeException("Exception while loading the forest : "
+ + e.getMessage());
+ }
+
+ configure(p, dataset, keys, trees, numInstances);
+ }
+
+ /**
+ * Compute the number of trees that need to classify the instances of this
+ * mapper's partition
+ *
+ * @param numMaps total number of map tasks
+ * @param numTrees total number of trees in the forest
+ * @param partition mapper's partition
+ * @return
+ */
+ public static int nbConcerned(int numMaps, int numTrees, int partition) {
+ assert partition >= 0 : "partition < 0";
+ // the trees of the mapper's partition are not concerned
+ return numTrees - Step1Mapper.nbTrees(numMaps, numTrees, partition);
+ }
+
+ /**
+ * Useful for testing. Configures the mapper without using a JobConf<br>
+ * TODO we don't need the keys partitions, the tree ids should suffice
+ *
+ * @param partition mapper's partition
+ * @param dataset
+ * @param keys keys returned by the first step
+ * @param trees trees returned by the first step
+ * @param numInstances number of instances in the mapper's partition
+ */
+ public void configure(int partition, Dataset dataset, TreeID[] keys,
+ Node[] trees, int numInstances) {
+ this.partition = partition;
+ if (partition < 0) {
+ throw new RuntimeException("Wrong partition id : " + partition);
+ }
+
+ converter = new DataConverter(dataset);
+
+ assert keys.length == trees.length : "keys.length != trees.length";
+ int nbConcerned = keys.length;
+
+ this.keys = keys;
+ this.trees = trees;
+
+ // make sure the trees are not from this partition
+ for (TreeID key : keys) {
+ assert key.partition() != partition : "a tree from this partition was found !";
+ }
+
+ // init the callbacks
+ callbacks = new SingleTreePredictions[nbConcerned];
+ for (int index = 0; index < nbConcerned; index++) {
+ callbacks[index] = new SingleTreePredictions(numInstances);
+ }
+
+ }
+
+
+ @Override
+ protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+
+ Instance instance = converter.convert(instanceId, value.toString());
+
+ for (int index = 0; index < keys.length; index++) {
+ int prediction = trees[index].classify(instance);
+ callbacks[index].prediction(index, instanceId, prediction);
+ }
+
+ instanceId++;
+ }
+
+
+ @Override
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ for (int index = 0; index < keys.length; index++) {
+ TreeID key = new TreeID(partition, keys[index].treeId());
+ context.write(key, new MapredOutput(callbacks[index].predictions));
+ }
+ }
+
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/TreeID.java
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/TreeID.java?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/TreeID.java (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/TreeID.java Wed Sep 30 05:29:22 2009
@@ -0,0 +1,57 @@
+/**
+ * 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.mapreduce.partial;
+
+import org.apache.hadoop.io.LongWritable;
+
+/**
+ * Indicates both the tree and the data partition used to grow the tree
+ */
+public class TreeID extends LongWritable {
+
+ public static final int MAX_TREEID = 100000;
+ public TreeID() {
+ }
+
+ public TreeID(int partition, int treeId) {
+ assert partition >= 0 : "partition < 0";
+ assert treeId >= 0 : "treeId < 0";
+ set(partition, treeId);
+ }
+
+ public void set(int partition, int treeId) {
+ super.set((long) partition * MAX_TREEID + treeId);
+ }
+
+ /**
+ * Data partition (InputSplit's index) that was used to grow the tree
+ * @return
+ */
+ public int partition() {
+ return (int) (get() / MAX_TREEID);
+ }
+
+ public int treeId() {
+ return (int) (get() % MAX_TREEID);
+ }
+
+ @Override
+ public TreeID clone() {
+ return new TreeID(partition(), treeId());
+ }
+}
Added: lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/package.html
URL: http://svn.apache.org/viewvc/lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/package.html?rev=820181&view=auto
==============================================================================
--- lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/package.html (added)
+++ lucene/mahout/trunk/core/src/main/java/org/apache/mahout/df/mapreduce/partial/package.html Wed Sep 30 05:29:22 2009
@@ -0,0 +1,41 @@
+<HTML>
+ <!--
+ /**
+ * 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.
+ */
+
+ --><HEAD>
+ <TITLE>org.apache.mahout.rf.mapred.partial</TITLE>
+</HEAD>
+<BODY>
+<DIV align="center">
+Partial-data mapreduce implementation of <b>Random Decision Forests</b>
+</DIV>
+<DIV> </DIV>
+<DIV>
+The builder splits the data, using a <b>FileInputSplit</b>, among the mappers. Building the forest and estimating the oob error takes two job steps.<br>
+<br>
+In the first step, each mapper is responsible for growing a number of trees with its partition's, loading the data instances in its map() function, then building the trees in the close() method. It uses the reference implementation's code to build each tree and estimate the oob error.<br>
+<br>
+The second step is needed when estimating the oob error. Each mapper loads all the trees that does not belong to its own partition (were not built using the partition's data) and uses them to classify the partition's data instances. The data instances are loaded in the map() method and the classification is performed in the close() method.<br>
+<br>
+</DIV>
+<DIV> </DIV>
+<DIV align="center">
+Copyright © 2009 <A HREF="http://lucene.apache.org/mahout">Apache Software Foundation - Mahout</A>
+</DIV>
+</BODY>
+</HTML>