Today, I decided to write a demo on how to implement customwritable ?
Let's say we have a dataset like following:
We want to calculate number of accidents on each day of week in each year. The output looks like :
As you can see the reducer is emitting key which is the combination of year and day in the sample data. This is where we need 'WritableComparable' interface.
In the mapper code, I am splitting the value on ',' and getting the year part out of date column. Also, I need day column from dataset . Once I have both keys, I am writing the context as :
context.write(new DOW(new Text(Integer.toString(year)), new Text(Integer.toString(day))),
new IntWritable(1));
Key -> new DOW(new Text(Integer.toString(year)), new Text(Integer.toString(day))
Value -> new IntWritable(1)
And that is it .
Happy Hadooping !!
Varun
Let's say we have a dataset like following:
AccidentId, Date, DayHere Day 5 means Thursday, 6 means Friday, 1 means Sunday and so on.
1, 1/1/1979, 5
2, 1/2/1979, 6
3, 1/3/1979, 7
4, 1/4/1979, 1
5, 1/5/1979, 2
6, 1/6/1979, 3
7, 1/7/1979, 4
8, 1/8/1979, 5
......
.......
16, 1/6/1980, 3
17, 1/7/1980, 4
18, 1/8/1980, 5
We want to calculate number of accidents on each day of week in each year. The output looks like :
| Number of accidents per day in the year 1979 |
As you can see the reducer is emitting key which is the combination of year and day in the sample data. This is where we need 'WritableComparable' interface.
public class DOW implements WritableComparable<DOW> {
private Text year;
private Text day;
// private int count;
public DOW() {
this.year = new Text();
this.day = new Text();
// this.count = count;
}
public DOW(Text year, Text day) {
this.year = year;
this.day = day;
// this.count = count;
}
public Text getYear() {
return this.year;
}
public void setYear(Text year) {
this.year = year;
}
public Text getDay() {
return this.day;
}
public void setDay(Text day) {
this.day = day;
}
@Override
public void readFields(DataInput in) throws IOException {
// TODO Auto-generated method stub
year.readFields(in);
day.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
// TODO Auto-generated method stub
year.write(out);
day.write(out);
}
@Override
public int compareTo(DOW o) {
// TODO Auto-generated method stub
int cmp = year.compareTo(o.year);
if (cmp != 0) {
return cmp;
}
return day.compareTo(o.day);
}
@Override
public String toString() {
// TODO Auto-generated method stub
return year + "," + day;
}
@Override
public int hashCode() {
// TODO Auto-generated method stub
return year.hashCode() * 163 + day.hashCode();
}
}
Once we have implemented the WritableComparable interface , we will use the class in Mapper and reducer code as follows.In the mapper code, I am splitting the value on ',' and getting the year part out of date column. Also, I need day column from dataset . Once I have both keys, I am writing the context as :
context.write(new DOW(new Text(Integer.toString(year)), new Text(Integer.toString(day))),
new IntWritable(1));
Key -> new DOW(new Text(Integer.toString(year)), new Text(Integer.toString(day))
Value -> new IntWritable(1)
protected void map(LongWritable key, Text value, Context context)Reducer is simply calculating the number of accidents based on key.
throws java.io.IOException, InterruptedException {
if (value.toString().contains(",")) {
String[] array = value.toString().split(",");
if (!array[9].equals("Date")) {
Date dt = null;
try {
dt = new SimpleDateFormat("dd/mm/yyyy").parse(array[9]);
} catch (ParseException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
int year = dt.getYear();
int day = Integer.parseInt(array[10].toString());
sLogger.info("year=" + year);
sLogger.info("array[10]=" + array[10]);
context.write(new DOW(new Text(Integer.toString(year)),
new Text(Integer.toString(day))),
new IntWritable(1));
}
}
@OverrideOnce you run this code, you will be able to see the output based on the key year and day.
protected void reduce(DOW key, Iterable<IntWritable> values,
Context context) throws java.io.IOException,
InterruptedException {
int count = 0;
sLogger.info("key =" + key);
for (IntWritable x : values) {
int val = Integer.parseInt(x.toString());
count = count + val;
}
context.write(key, new IntWritable(count));
};
And that is it .
Happy Hadooping !!
Varun
Good blog easy to understand, if possible please put in the interview questions faced by u in the blog
ReplyDelete