Monthly Archives: October 2014

The fomular of MapReduce can be described as follow. Besides, you can find the configuration of a Job.
map: (K1, V1) → list(K2, V2)
reduce: (K2, list(V2)) → list(K3, V3)

(K1, V1):
jobConf.setInputKeyClass(K1. class );
jobConf.setInputValueClass(V1. class );

list(K2, V2):
job.setMapOutputKeyClass(K2.class);
job.setMapOutputValueClass(V2.class);

list(K3, V3):
jobConf.setOutputKeyClass(K3. class );
jobConf.setOutputValueClass(V3. class );

Normally, (K2, V2) equals (K3, V3). I show the code1:

1. public class Dedup2 {
2.     public static class Map extends Mapper<Object,Text,Text,Text>{
3.         private static Text line=new Text();
4.         public void map(Object key,Text value,Context context)
5.                 throws IOException,InterruptedException{
6.             line=value;
7.             context.write(line, new Text(“”));
8.         }
9.     }
10.     public static class Reduce extends Reducer<Text,Text,Text,Text>{
11.         public void reduce(Text key,Iterable<Text> values,Context context)
12.                 throws IOException,InterruptedException{
13.             context.write(key, new Text(“”));
14.         }
15.     }
16.     public static void main(String[] args) throws Exception{
17.         Configuration conf = new Configuration();
18.         System.setProperty(“HADOOP_USER_NAME”, “root”);
19.         String[] otherArgs = {“hdfs://centmaster:9000/input”, “hdfs://centmaster:9000/output/debup1”};
20.         Job job = new Job(conf, “Data Deduplication”);
21.         job.setJarByClass(Dedup.class);
22.         job.setMapperClass(Map.class);
23.         job.setCombinerClass(Reduce.class);
24.         job.setReducerClass(Reduce.class);
25.         job.setOutputKeyClass(Text.class);
26.         job.setOutputValueClass(Text.class);
27.         FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
28.         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
29.         System.exit(job.waitForCompletion(true) ? 0 : 1);
30.      }
31. }

But I want to set a different (K2, V2) from (K3, V3). So I did the following change:

1. public class Dedup{
2.     public static class Map extends Mapper<Object,Text,Text,IntWritable>{   //Change Text to IntWritable
3.         private static Text line=new Text();
4.         public void map(Object key,Text value,Context context)
5.                 throws IOException,InterruptedException{
6.             line=value;
7.             context.write(line, new IntWritable(0));    //Change Text(“”) to IntWritable(0)
8.         }
9.     }
10.     public static class Reduce extends Reducer<Text,IntWritable,Text,Text>{     //Change Text to IntWritable
11.         public void reduce(Text key,Iterable<IntWritable> values,Context context)   //Change Text to IntWritable
12.                 throws IOException,InterruptedException{
13.             context.write(key, new Text(“”));
14.         }
15.     }
16.     public static void main(String[] args) throws Exception{
17.         Configuration conf = new Configuration();
18.         System.setProperty(“HADOOP_USER_NAME”, “root”);
19.         String[] otherArgs = {“hdfs://centmaster:9000/input”, “hdfs://centmaster:9000/output/debup2”};
20.         Job job = new Job(conf, “Data Deduplication”);
21.         job.setJarByClass(Dedup.class);
22.         job.setMapperClass(Map.class);
23.         job.setCombinerClass(Reduce.class);
24.         job.setReducerClass(Reduce.class);
25.         job.setOutputKeyClass(Text.class);
26.         job.setOutputValueClass(Text.class);
27.     job.setMapOutputValueClass(IntWritable.class);    //New added. To indicate the Mapper Output Value is not default Text, but IntWritable
28.         FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
29.         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
30.         System.exit(job.waitForCompletion(true) ? 0 : 1);
31.      }
32. }

After the change, I run the code. It doesn’t generate the output. I found out that Reducer is not running. There is no error information in eclipse. I added the try…catch… to Reducer code:

1.     public static class Reduce extends Reducer<Text,IntWritable,Text,Text>{
2.         public void reduce(Text key,Iterable<IntWritable> values,Context context)
3.                 throws IOException,InterruptedException{
4.             try{
5.                 context.write(key, new Text(“”));
6.                 System.out.println(“in reducer”);
7.             }catch(Exception e){
8.                 System.out.println(e);
9.             }
10.         }
11.     }

The error information shows:
java.io.IOException: wrong value class: class org.apache.hadoop.io.Text is not class org.apache.hadoop.io.IntWritable

Google it, I found it is because that the Combiner shouldn’t be set. So I commented this line:
//job.setCombinerClass(Reduce.class);

Ding… The output came out. This debug took me 1 day to solve.

Summary:
Back to the original rule of Mapper and Reducer. It is like below:

(K1, V1):
jobConf.setInputKeyClass(K1. class );
jobConf.setInputValueClass(V1. class );

list(K2, V2):
job.setMapOutputKeyClass(K2.class);
job.setMapOutputValueClass(V2.class);

list(K3, V3):
jobConf.setOutputKeyClass(K3. class );
jobConf.setOutputValueClass(V3. class );

But if you want a different (K2, V2) from (K3, V3), the CombinerClass shouldn’t be set. Comment it.

I have already deployed 4 nodes hadoop cluster in VMWare. I will run the wordcount application in myeclipse. Myeclipse is run in win8, which is my own laptop.
1. Deploy a hadoop work environment in eclipse. (You can refer to “Build hadoop work environment in MyEclipse” in my blog).
2. Import hadoop-mapreduce-client-common-2.3.0.jar and hadoop-mapreduce-client-jobclient-2.3.0.jar to workspace. They exist in HADOOP/share/hadoop/mapreduce directory.

3. You should still have a hadoop directory in windows. And set set the environment vairable of HADOOP_HOME. This is mandantory.

4. Download hadoop-common-2.2.0-bin-master.rar, extract the files and overwrite in hadoop/bin. Here, there are 2 necessary files. They are hadoop.dll and winutils.exe.

6. I copied file1.txt, file2.txt to centmaster:9000/input

5. Modify the WordCount.java, and run it in eclipse.
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {

public static class TokenizerMapper extends
Mapper《Object, Text, Text, IntWritable》 {

private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}

public static class IntSumReducer extends
Reducer《Text, IntWritable, Text, IntWritable》 {
private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = { “hdfs://centmaster:9000/input”,
“hdfs://centmaster:9000/output” };
System.setProperty(“HADOOP_USER_NAME”, “root”);        //this is very important. Set the root name of the NameNode(centmaster).
Job job = new Job(conf, “word count”);
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
6. You can find the result.


Use the cat to check the result:
[root@centmaster bin]# bash hadoop fs -cat /output/part-r-00000
2012-3-1        2
2012-3-2        2
2012-3-3        4
2012-3-4        2
2012-3-5        2
2012-3-6        2
2012-3-7        2
a       4
b       4
c       5
d       3

Some difference from the online articles.
1. Some articles said, you should configure the hadoop/etc/hadoop/hadoop-env.cmd, and set the JAVA_HOME. But in my case, it is not necessary. I tested, that I modified “etc” into “etc2”. Mapreduce runs well.
2. Some articles said, that you should put hadoop.dll in windows/system32. I copied, but I didn’t test if it is necessary to do that. You can try that. But you need to restart your computer.
3. I didn’t configure the %HADOOP_HOME%\bin in PATH environment viariable, it still runs well.

Experience to debug:
Eclipse doesn’t show many error information. The best way is that you find it in logs/hadoop-root-namenode-centmaster.log file in the namenode, and google it.