Thursday 24 December 2015
Input:
Program:
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
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.Reducer;
import org.apache.hadoop.mapred.Reporter;
public class MarketRating {
public static class MapClass extends MapReduceBase implements
Mapper<LongWritable, Text, Text, Text> {
private Text loc = new Text();
private Text rating = new Text();
@Override
public void map(LongWritable key, Text value,
OutputCollector<Text, Text> output, Reporter reporter)
throws IOException {
String[] rows = value.toString().split(",");
if (rows.length > 31) {
String city = rows[4];
String state = rows[6];
int count = 0;
int rated = 0;
for (int col = 11; col <= 31; col++) // columns 11-31 contain
// data about what the
// market offers
{
if (rows[col].equals("Y"))
count++;
}
count = (count * 100) / 21; // gets 1-100 rating of the market
if (count > 0) {
rated = 1;
}
loc.set(city + ", " + state);
rating.set(1 + "\t" + rated + "\t" + count); // numTotal,
// numRated,
// rating
output.collect(loc, rating);
}
}
}
public static class Reduce extends MapReduceBase implements
Reducer<Text, Text, Text, Text> {
@Override
public void reduce(Text key, Iterator<Text> values,
OutputCollector<Text, Text> output, Reporter reporter)
throws IOException {
int rating = 0;
int numRated = 0;
int numTotal = 0;
while (values.hasNext()) {
String tokens[] = (values.next().toString()).split("\t");
int tot = Integer.parseInt(tokens[0]);
int num = Integer.parseInt(tokens[1]); // gets number of markets
int val = Integer.parseInt(tokens[2]); // gets rating
if (val > 0) // filters out markets with no data
{
rating = (rating * numRated + val * num) / (numRated + num);
numRated = numRated + num;
}
numTotal = numTotal + tot;
}
if (rating > 0)
output.collect(key, new Text(numTotal + "\t" + numRated + "\t"
+ rating));
}
}
public static void main(String[] args) throws IOException {
JobConf conf = new JobConf(MarketRating.class);
conf.setJobName("MarketRatings");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Text.class);
conf.setMapperClass(MapClass.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
JobClient.runJob(conf);
}
}
Output:
Very nicely explained .
ReplyDeletemytectra
Nice article, Thanks for sharing such a nice information.
ReplyDeleteTry , Peoplesoft Admin Training
This comment has been removed by a blog administrator.
ReplyDelete