A few weeks ago I had a need for total order sorting, and I couldn't
find any easy to use, general purpose code for getting the partition
file setup.
Attached is the code I settled on. Would it be worthwhile to add to
hadoop-mapred?
My goals were:
1) Extract all needed setup from an existing configuration
2) Run quickly
3) Easy to use
It accepts a single Configuration, which should be a ready to run job.
It then changes the input format and reducer to custom classes.
The sampling inputformat uses the original inputformat's record
reader, but makes the input appear much smaller to the mapper
(1/1000th by default).
Those inputs go through the original mapper, and that mapper's output
is sent to a single partition.
The original reducer is replaced with a simple sampling reducer that
only emits mapred.reduce.tasks-1 records.
It reads the entire input set, so that should go fast. Also, the
mapper shouldn't have any side effects, like inserting into HBase,
since the prep job uses the original.
I took this approach because I have many jobs where the mapper changes
the key type from what the InputFormat provides. It also made the
sampler independent of the job's input type.
I don't have any example programs, but if there's interest, I can provide some.
Thanks,
Chase Bradford
--
“If in physics there's something you don't understand, you can always
hide behind the uncharted depths of nature. But if your program
doesn't work, there is no obstinate nature. If it doesn't work, you've
messed up.”
- Edsger Dijkstra
package org.apache.hadoop.mapreduce.lib.partition;
import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.UUID;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.util.ReflectionUtils;
/**
* This class provides a static apply method which uses a Configuration to
* run a sampling job that sets up a partition file to use in the TotalOrderPartition
*/
public class TotalOrderPrep {
private static final String ARG_SAMPLESIZE = "top.sample.size";
private static final String ARG_INPUTRATE = "top.input.rate";
private static final String ARG_INPUTFORMAT = "top.inputformat.class";
public static void apply(Configuration conf) {
Configuration prep = new Configuration(conf);
// If there's only one reduce task, don't bother with total order
// partitioning.
if( conf.getInt("mapred.reduce.tasks", 1) == 1 ) {
conf.set("mapreduce.partitioner.class", HashPartitioner.class.getName());
return;
}
// Remember the real input format so the sampling input format can use
// it under the hood
prep.setClass(ARG_INPUTFORMAT, conf.getClass("mapreduce.inputformat.class", TextInputFormat.class));
prep.setClass("mapreduce.inputformat.class", SamplerInputFormat.class, InputFormat.class);
// Base the sample size on the number of reduce tasks that will be used
// by the real job, but only use 1 reducer for this job.
prep.set(ARG_SAMPLESIZE, conf.get("mapred.reduce.tasks"));
prep.setInt("mapred.reduce.tasks", 1);
// Make this job's output the input file for the real job's
// TotalOrderPartitioner
String partition = "/tmp/topPartition/" + UUID.randomUUID();
prep.set("mapred.output.dir", partition);
conf.set("total.order.partitioner.path", partition + "/part-r-00000");
conf.set("mapreduce.partitioner.class", TotalOrderPartitioner.class.getName());
// The final output key must match the map output key
if( null != prep.get("mapred.mapoutput.key.class") )
prep.set("mapred.output.key.class", conf.get("mapred.mapoutput.key.class"));
// The map output value type must not change, which would happen if it's
// not explicitly defined, and we change the job's final output to
// NullWritable
if( null == prep.get("mapred.mapoutput.value.class") ) {
Class outvalue = conf.getClass("mapred.output.value.class", Text.class);
prep.setClass("mapred.mapoutput.value.class", outvalue, Writable.class);
}
prep.set("mapred.output.value.class", NullWritable.class.getName());
// No need for a real partitioner (only one part), so just make sure it's
// a dependency free one.
prep.set("mapreduce.partitioner.class", HashPartitioner.class.getName());
prep.set("mapreduce.reduce.class", SamplingReducer.class.getName());
prep.set("mapreduce.outputformat.class", SequenceFileOutputFormat.class.getName());
prep.set("mapred.job.name", conf.get("mapred.job.name", "Unknown Job") + " (Sampler)");
// Run the job. If it fails, then the original job probably has an error.
try {
Job job = new Job(prep);
job.setJarByClass(TotalOrderPrep.class);
job.waitForCompletion(false);
if( !job.isSuccessful() )
throw new RuntimeException("Partition sampler job failed.");
} catch (Exception e) {
throw new RuntimeException("Failed to start Partition sampler.", e);
}
}
/**
* An input format that makes a record reader which passes a subset of the
* KV pairs to produced by the underlying record reader
*/
public static class SamplerInputFormat<K,V> extends FileInputFormat<K, V> {
@Override
public RecordReader<K, V> createRecordReader( InputSplit split, TaskAttemptContext context)
throws IOException, InterruptedException {
Configuration conf = context.getConfiguration();
Class inputformat = conf.getClass(ARG_INPUTFORMAT, TextInputFormat.class);
int rate = conf.getInt(ARG_INPUTRATE, 1000);
@SuppressWarnings("unchecked")
InputFormat<K,V> realif = (InputFormat<K,V>) ReflectionUtils.newInstance(inputformat, conf);
return new SamplingRecordReader<K,V>(realif.createRecordReader(split, context), rate);
}
}
/**
* A record reader which only passes through a reduced set of KV pairs.
*/
public static class SamplingRecordReader<K,V> extends RecordReader<K, V> {
private final int rate;
private final RecordReader<K,V> real;
SamplingRecordReader(RecordReader<K,V> real, int rate) {
this.real = real;
this.rate = rate;
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
// Make the underlying RR skip many records before emitting to our caller
int n = this.rate;
boolean result = real.nextKeyValue();
while( n > 0 && result ) {
--n;
result = real.nextKeyValue();
}
return result;
}
@Override
public void close() throws IOException {
real.close();
}
@Override
public K getCurrentKey() throws IOException, InterruptedException {
return real.getCurrentKey();
}
@Override
public V getCurrentValue() throws IOException, InterruptedException {
return real.getCurrentValue();
}
@Override
public float getProgress() throws IOException, InterruptedException {
return real.getProgress();
}
@Override
public void initialize(InputSplit split, TaskAttemptContext context)
throws IOException, InterruptedException {
real.initialize(split, context);
}
}
/**
* This reducer accumulates keys into a fixed sized buffer that evenly span
* the entire key space.
*/
public static class SamplingReducer<K1,V1,K2,V2> extends Reducer<K1, V1, K1, NullWritable> {
private static final int BUFFER_MAX = 1000;
@SuppressWarnings("unchecked")
private LinkedList<K1> keys = new LinkedList();
private int target;
// Whenever the buffer fills up, drop every element in an even index
// then collect at only half the frequency
private int collect_rate = 1;
private int collect = 1;
@Override
@SuppressWarnings("unchecked")
protected void reduce(K1 key, Iterable<V1> value, Context context)
throws IOException, InterruptedException {
--collect;
if( collect == 0 ) {
keys.add( (K1) WritableUtils.clone((Writable)key, context.getConfiguration()) );
collect = collect_rate;
}
if( keys.size() < BUFFER_MAX)
return;
// Down sample by half, then double the collection rate.
Iterator<K1> i = keys.iterator();
while( i.hasNext() ) {
i.next();
i.remove();
if( i.hasNext() ) i.next();
}
collect_rate *= 2;
collect = collect_rate;
}
// Flush the collected keys
// What's buffered should span the key space pretty evenly, but we still
// only need an even sampling of those.
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
float rate = keys.size() / (float)target;
float emit = 0;
for( K1 key : keys ) {
emit += 1.0f;
if( emit > rate ) {
context.write(key, NullWritable.get());
emit -= rate;
}
}
}
@Override
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
target = context.getConfiguration().getInt(ARG_SAMPLESIZE, 1);
}
}
}
