I'm really new to MapReduce and currently struggling with a MapReduce exercise. I'm given a really big dataset of U.S. patents granted between January 1963 and December 1999 that looks something like this:
"PATENT","GYEAR","GDATE","APPYEAR","COUNTRY","POSTATE","ASSIGNEE","ASSCODE","CLAIMS","NCLASS","CAT","SUBCAT","CMADE","CRECEIVE","RATIOCIT","GENERAL","ORIGINAL","FWDAPLAG","BCKGTLAG","SELFCTUB","SELFCTLB","SECDUPBD","SECDLWBD"
3070801,1963,1096,,"BE","",,1,,269,6,69,,1,,0,,,,,,,
3070802,1963,1096,,"US","TX",,1,,2,6,63,,0,,,,,,,,,
3695459,1972,4659,1970,"BE","",225495,3,,414,5,51,,2,,0.5,,16.5,,,,0,0
3695460,1972,4659,1970,"IT","",,1,,414,5,51,,1,,0,,28,,,,,
3866063,1975,5520,1973,"US","CA",188385,2,5,327,4,41,7,11,1,0.7438,0.4082,10.8182,5,0,0,0,0
3866064,1975,5520,1973,"US","FL",242085,2,10,327,4,41,5,11,1,0.314,0,10.9091,3.6,0,0,0,0
4548215,1985,9426,1983,"DE","",243670,3,25,131,6,61,2,7,1,0.2449,0,5.8571,19.5,0,0,0,0
4548216,1985,9426,1982,"GB","",382615,3,7,131,6,61,10,3,0.6,0,0.2778,5,20,0.5,0.2,0,0
...
These are the meanings of some attributes:
I need to implement a MapReduce Java class (which includes a Partitioner) to compute the average number of claims per patent grouped by country since 1975. The Partitioner sorts the results into two groups: The result for AU needs to be in the 1st group & those for other countries in the 2nd group. I will then need to show the result for AU using the Hadoop command.
This is my code:
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class AvgClaimNumbersPerPatentByCountry extends Configured implements Tool {
public int run(String[] args) throws Exception {
Configuration conf = this.getConf();
Job job = Job.getInstance(conf, "Avg Claims/Patent by Country since 1975");
job.setJarByClass(AvgClaimNumbersPerPatentByCountry.class);
job.setMapperClass(MyMapper.class); //the Mapper class
job.setReducerClass(MyReducer.class); //the Reducer class
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setOutputValueClass(FloatWritable.class);
job.setCombinerClass(MyReducer.class); //the Combiner class
job.setPartitionerClass(MyPartitioner.class); //the Partitioner class
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
return job.waitForCompletion(true) ? 0 : 1;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new AvgClaimNumbersPerPatentByCountry(), args);
System.exit(res);
}
public static class MyMapper extends Mapper<Object, Text, Text, IntWritable> {
private Text countryAsKey = new Text();
private IntWritable claimsAsValue = new IntWritable(1);
//map() function vvv
@Override
public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
if (line.contains("PATENT")) {
return; //skip the first line
}
else {
String[] patentData = line.split(",");
countryAsKey.set(patentData[4]);
if (Integer.parseInt(patentData[1]) >= 1975) {
claimsAsValue.set(Integer.parseInt(patentData[8]));
}
}
context.write(countryAsKey, claimsAsValue);
}
//map() function ^^^
}
public static class MyPartitioner extends Partitioner<Text, IntWritable> {
//getPartition() function vvv
public int getPartition(Text key, IntWritable value, int numReduceTasks) {
String country = key.toString();
if (country.toLowerCase().matches("AU")) {
return 0;
}
else {
return 1;
}
}
//getPartition() function ^^^
}
public static class MyReducer extends Reducer<Text, IntWritable, Text, FloatWritable> {
//reduce() function vvv
@Override
public void reduce(Text countryKey, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int totalClaimsOfCountry = 0;
int totalPatentsOfCountry = 0;
FloatWritable avgClaim = new FloatWritable();
for (IntWritable value : values) {
totalClaimsOfCountry += value.get();
totalPatentsOfCountry += 1;
}
avgClaim.set(calculateAvgClaimPerPatent(totalClaimsOfCountry, totalPatentsOfCountry));
context.write(countryKey, avgClaim);
}
//reduce() function ^^^
public float calculateAvgClaimPerPatent(int totalClaims, int totalPatents) {
float avg = (float)totalClaims / totalPatents;
return avg;
}
}
}
When I run the MapReduce application in Hadoop, I got this error:
21/08/19 01:18:50 INFO mapreduce.Job: Task Id : attempt_1629275188478_0001_m_000001_0, Status : FAILED Error: java.io.IOException: Type mismatch in value from map: expected org.apache.hadoop.io.FloatWritable, received org.apache.hadoop.io.IntWritable
I understand that there are mismatched values in the code (I passed in FloatWritable while the system expected IntWritable) but I'm not sure how to fix this. I would really appreciate any ideas and help. Please also point out if there is anything wrong with my code so that I can improve in my future exercises :)
Thanks a lot in advance!