Hadoop 의 기본예제인 WordCount 를 기반으로 설명한다.
프로그래밍은 아래와 같은 4가지 부분에 대하여 작성한다.
기본 분산처리된 데이터는 각 네임노드의 mapper 와 combiner 의 context에 저장된 값이 reducer 로 취합되며
reduce 취합시에는 key 로 정렬된다.
흐름은 아래와 같다.
Job 등록 및 시작 ↓
Mapper ↓ Combiner ↓ shuffle ↓ Reducer ↓ 최종결과 hdfs에 저장 |
Mapper 작성
Mapper를 작성시에는 Mapper <KEYIN, VALUEIN, KEYOUT, VALUEOUT> 클래스의 KEYIN, VALUEIN, KEYOUT, VALUEOUT의 4가지를 정의 하여 상속한다. Map 의 경우에는 KEYIN key, VALUEIN value, Context context 의 3가지 파라미터로 구현하며, Type 은 클래스에서 정의 한 타입과 동일하게 구현한다.
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); } } } |
Mapper 구현 부분의 map 함수가 파일을 TextInputFormat 형태로, 데이터의 한줄(line)씩 읽은 값이 파리미터 Text value 로 받는다. 이 값을StringTokenizer 를 이용하여, 줄 (line) 을 token 으로 나누고 key-value pair 인 <word, 1> 를 Combiner 또는 Reducer 에 전달한다.
Combiner 작성
Combiner 는 Reduce 의 동일한 ReduceContext <KEYIN, VALUEIN, KEYOUT, VALUEOUT>를 상속하여 구현한다.
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); } } |
WordCount 의 combine를 정의하며. map 의 결과물 (output) 은 local combiner 에게 전달되어 정렬되고, 합산되어 Reducer 에 전달된다.
Reducer 작성
Reduce 의 동일한 ReduceContext <KEYIN, VALUEIN, KEYOUT, VALUEOUT>를 상속하여 구현한다.
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); } } |
Reducer 구현 부분의 reduce 함수는 각 key 별로 shuffle되어 같은 값으로 모아진 Iterable<IntWritable> values 를 카운팅하여 해당 Text 값을 key 로 개수를 최종값으로 저장한다.
Job 등록
위와 같은 방식으로 작성된 Mapper, Combiner, Reducer 는 Job 으로 등록하여 실행한다.
public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(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(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } } |
run 함수는 input/output 경로나, key-value 의 타입, input/output 의 포맷등을 JobConf 를 통해서 job 의 여러 성질을 정의한다.그리고 JobClient.runJob 을 호출하여 job 을 실행하고 진행상황을 모니터한다.
이 경우에는 combiner 와 reducer 에 같은 클래스를 등록했다. 두 클래스를 다루는 자세한 내용은 이후에 다루도록 하겠다