Commit 87059a12 authored by feng's avatar feng

Stripes 1

parent 392e2ab0
This diff is collapsed.
......@@ -2,7 +2,14 @@ package fr.eurecom.dsg.mapreduce;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
/*
* Very simple (and scholastic) implementation of a Writable associative array for String to Int
......@@ -11,12 +18,28 @@ import org.apache.hadoop.io.Writable;
public class StringToIntMapWritable implements Writable {
// TODO: add an internal field that is the real associative array
private HashMap<Text, IntWritable> hashMap;
public StringToIntMapWritable() {
hashMap = new HashMap<Text, IntWritable>();
}
public StringToIntMapWritable(Text text, IntWritable intWritable) {
this();
hashMap.put(text, intWritable);
}
@Override
public void readFields(DataInput in) throws IOException {
// TODO: implement deserialization
Text text = new Text();
IntWritable intWritable = new IntWritable();
text.readFields(in);
intWritable.readFields(in);
hashMap.put(text, intWritable);
// Warning: for efficiency reasons, Hadoop attempts to re-use old instances of
// StringToIntMapWritable when reading new records. Remember to initialize your variables
// inside this function, in order to get rid of old data.
......@@ -27,5 +50,18 @@ public class StringToIntMapWritable implements Writable {
public void write(DataOutput out) throws IOException {
// TODO: implement serialization
Set<Text> keys = hashMap.keySet();
Iterator iterator = keys.iterator();
if (iterator.hasNext()) {
// write key
Text text = (Text) iterator.next();
text.write(out);
// write value
IntWritable intWritable = hashMap.get(iterator.next());
intWritable.write(out);
}
}
}
package fr.eurecom.dsg.mapreduce;
import java.io.IOException;
import java.util.LinkedList;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
......@@ -14,70 +18,90 @@ import org.apache.hadoop.util.ToolRunner;
public class Stripes extends Configured implements Tool {
private int numReducers;
private Path inputPath;
private Path outputDir;
@Override
public int run(String[] args) throws Exception {
Configuration conf = this.getConf();
Job job = null; // TODO: define new job instead of null using conf e setting a name
// TODO: set job input format
// TODO: set map class and the map output key and value classes
// TODO: set reduce class and the reduce output key and value classes
// TODO: set job output format
// TODO: add the input file as job input (from HDFS) to the variable inputFile
// TODO: set the output path for the job results (to HDFS) to the variable outputPath
// TODO: set the number of reducers using variable numberReducers
// TODO: set the jar class
return job.waitForCompletion(true) ? 0 : 1;
}
public Stripes (String[] args) {
if (args.length != 3) {
System.out.println("Usage: Stripes <num_reducers> <input_path> <output_path>");
System.exit(0);
private int numReducers;
private Path inputPath;
private Path outputDir;
@Override
public int run(String[] args) throws Exception {
Configuration conf = this.getConf();
Job job = null; // TODO: define new job instead of null using conf e setting a name
// TODO: set job input format
// TODO: set map class and the map output key and value classes
// TODO: set reduce class and the reduce output key and value classes
// TODO: set job output format
// TODO: add the input file as job input (from HDFS) to the variable inputFile
// TODO: set the output path for the job results (to HDFS) to the variable outputPath
// TODO: set the number of reducers using variable numberReducers
// TODO: set the jar class
return job.waitForCompletion(true) ? 0 : 1;
}
public Stripes(String[] args) {
if (args.length != 3) {
System.out.println("Usage: Stripes <num_reducers> <input_path> <output_path>");
System.exit(0);
}
this.numReducers = Integer.parseInt(args[0]);
this.inputPath = new Path(args[1]);
this.outputDir = new Path(args[2]);
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new Stripes(args), args);
System.exit(res);
}
this.numReducers = Integer.parseInt(args[0]);
this.inputPath = new Path(args[1]);
this.outputDir = new Path(args[2]);
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new Stripes(args), args);
System.exit(res);
}
}
class StripesMapper
extends Mapper<Object, // TODO: change Object to input key type
Object, // TODO: change Object to input value type
Object, // TODO: change Object to output key type
Object> { // TODO: change Object to output value type
@Override
public void map(Object key, // TODO: change Object to input key type
Object value, // TODO: change Object to input value type
Context context)
throws IOException, InterruptedException {
// TODO: implement map method
}
extends Mapper<LongWritable, // TODO: change Object to input key type
Text, // TODO: change Object to input value type
Text, // TODO: change Object to output key type
StringToIntMapWritable> { // TODO: change Object to output value type
@Override
public void map(LongWritable key, // TODO: change Object to input key type
Text value, // TODO: change Object to input value type
Context context)
throws IOException, InterruptedException {
// TODO: implement map method
IntWritable ONE = new IntWritable(1);
String line = value.toString();
String[] words = line.split("\\s+");
for (String firstWord : words) {
LinkedList<StringToIntMapWritable> associativeArray = new LinkedList<StringToIntMapWritable>();
StringToIntMapWritable stripe;
for (int i = 0; i < words.length; i++) {
if (firstWord != words[i]) {
Text secondWord = new Text(words[i]);
stripe = new StringToIntMapWritable(secondWord, ONE);
if (associativeArray.contains(stripe)) {
} else {
associativeArray.add(stripe);
}
}
}
}
}
}
class StripesReducer
extends Reducer<Object, // TODO: change Object to input key type
Object, // TODO: change Object to input value type
Object, // TODO: change Object to output key type
Object> { // TODO: change Object to output value type
@Override
public void reduce(Object key, // TODO: change Object to input key type
Iterable<Object> values, // TODO: change Object to input value type
Context context) throws IOException, InterruptedException {
// TODO: implement the reduce method
}
extends Reducer<Text, // TODO: change Object to input key type
StringToIntMapWritable, // TODO: change Object to input value type
Object, // TODO: change Object to output key type
Object> { // TODO: change Object to output value type
@Override
public void reduce(Object key, // TODO: change Object to input key type
Iterable<Object> values, // TODO: change Object to input value type
Context context) throws IOException, InterruptedException {
// TODO: implement the reduce method
}
}
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment