hadoop wordcount 예제 분석.

기본적인 wordcount를 해주는 프로그램이다.
하지만 여기에는 wordcount 클래스가 존재하고 각각 상속받은 클래스에서 함수를 불러서 실행이 된다.
각각에 해당하는 api와 어떻게 쓰이는지 나중을 위해 써놓는다.
 

1.	package org.myorg;
2.
3.	import java.io.IOException;
4.	import java.util.*;
5.
6.	import org.apache.hadoop.fs.Path;
7.	import org.apache.hadoop.conf.*;
8.	import org.apache.hadoop.io.*;
9.	import org.apache.hadoop.mapred.*;
10.	import org.apache.hadoop.util.*;
11.
12.	public class WordCount {
13.
14.	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> { //이 부분은 기본적으로 써주는 map에서의 함수 인자값들. //LongWritable 은 long이고 text는 string이고 intWritable은 int이다.
15.	private final static IntWritable one = new IntWritable(1);
16.	private Text word = new Text();
17.
18.	public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
19.	String line = value.toString();//객체가 가진 값을 문자열로 출력.한 것을 line에 넣는다.
20.	StringTokenizer tokenizer = new StringTokenizer(line);//객체의 값을  한줄한줄 값을 가지고온다.
21.	while (tokenizer.hasMoreTokens()) {//끝이 될때까지 반복
22.	word.set(tokenizer.nextToken());//org.apache.hadoop.io 포함.(대부분의 text관련api들이 포함.
23.	output.collect(word, one);//<key,value>로 구성됨
24.	//이 부분은 하나의 줄을 가지고와 단어를 가지고 오는 역할을 한다고 생각하면 될거같다. }
25.	}
26.	}
27.
28.	public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
29.	public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
30.	int sum = 0;
31.	while (values.hasNext()) {//Iterator api에 있음 hasNext
32.	sum += values.next().get();//Iterator api에 있음 next()는 int값을 리턴.
33.	}
34.	output.collect(key, new IntWritable(sum)); //key와 value를 collect에 넣는다.
35.	}
36.	}
37.
38.	public static void main(String[] args) throws Exception {
39.	JobConf conf = new JobConf(WordCount.class); //job을 생성.
40.	conf.setJobName("wordcount");
41.
42.	conf.setOutputKeyClass(Text.class);//key는 text로
43.	conf.setOutputValueClass(IntWritable.class);//value는 int로
44.
45.	conf.setMapperClass(Map.class);//map을 실행.
46.	conf.setCombinerClass(Reduce.class);//테스크간에 데이터 전송을 최소할 필요가 있어서 컴바인 함수를 이용한다.(컴바이너 함수의 출력이 결국 리듀스 함수의 입력이 된다.)
47.	conf.setReducerClass(Reduce.class);
48.
49.	conf.setInputFormat(TextInputFormat.class);
50.	conf.setOutputFormat(TextOutputFormat.class);
51.
52.	FileInputFormat.setInputPaths(conf, new Path(args[0]));//입력될 폴더
53.	FileOutputFormat.setOutputPath(conf, new Path(args[1]));//출력될 위치 //이곳은 존재하면 안된다. 파일이 겹치게 되면 기존에 잇는게 날라가기 때문에 있어야된다.
54.
55.	JobClient.runJob(conf);
57.	}
58.	}