[Hadoop-03]맵리듀스 예제(WordCount)

Kamangs 2017. 8. 27. 02:45

2017. 8. 27. 02:45

728x90

하둡의 상징과도 같은 기술이다. 바로 맵리듀스이다.

사실 하둡만의 상징 같은건 아니고 데이터를 처리하는 솔루션에는 맵리듀스라는개념은 전부 존재한다.

예를들어 DB에도 맵리듀스라는 개념은 존재하는 것이다.

맵리듀스는 무엇인가? 사실 정말 간단한 개념이라서 설명하기 어렵진 않다.

간단히 말하면 맵리듀스는 맵과 리듀스로 나뉘며 맵은 특정 데이터를 가져와서 그걸 키와 밸류의 쌍으로 묶는다.

리듀스는 맵에서 묶은 키와 쌍을 이용해서 내가 필요한 정보로 다시 키와 쌍으로 묶게 된다.

즉 사실상 로직적으로 보면 맵이나 리듀스나 비슷비슷한 작업이다.

차이가 있다면 맵은 raw데이터, 즉 아무데이터나 받아와서 그걸 키와 밸류의 쌍으로 묶는 작업인것이고

리듀스는 이미 키와 밸류로 묶어진 데이터들을 다시 나에게 맞게 키와 밸류로 재조립하는 작업이라고 할 수 있다.

사실 백문이 불여일견이라고 이번예제를 통해서 맵리듀스를 매우 쉽게 이해할 수 있을 것이다.

이번 예제는 WordCount라는 하둡에서 굉장히 유명한 입문 예제이다.

이 예제를 통해서 맵리듀스를 통해서 알아보자.

맵리듀스의 구성

맵리듀스작업을 위해서는 반드시 3가지의 클래스가 필요하다.

이는 각각 맵과 리듀스 그리고 드라이버이다.

앞에서 맵과 리듀스는 설명을 했으므르로 드라이버가 먼놈인지 설명을 해야할거같은데

거창하게 말해서 드라이버지 그냥 자바를 실행시켜줄 main함수가 들은 놈을 드라이버라고 부른다.

맵리듀스의 구성은 정말 심플하지 않은가?

맵,리듀스,드라이버만 있으면 하둡프로그래밍을 할 수 있다고 말 할 수 있다.

이제 아래의 예제는 맵과 리듀스,드라이버를 하나씩 구현해 볼 것이다.

맵

package hadoop;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
            throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}

맵리듀스 프로그래밍에서 맵이라는것은 raw데이터를 받아와서 그것을 key와 value로 바꾸는 작업이라고 했다.

그러나 사실은 raw데이터조차도 key와 value로 가져온다.

일단 맵을 만들기 위해서는 매퍼클래스를 상속받아야된다.

매퍼클래스는 4개의 인자를 받는다.

여기서 맵리듀스에서의 인자는 일반 long형이나 int형, String형을 쓰지 않는다.

Long이나 int형은 각각 LongWritable, IntWritable이라는 하둡 전용 자료형을 쓴다.

그 이유는 전송을 가능하게 바이트별로 보내는 솔루션이 내장되어 있기 때문이다.

그리고 String은 Text로 사용한다. 이유는 위와같다.

매퍼클래스의 인자는 순서대로 인풋key,인풋value,아웃풋key,아웃풋value의 뜻을 가진다.

이게 뭘 의미하냐면 맵이라는 솔루션에 인풋으로 들어오는것 역시 key와 value쌍이고

나가는것 역시 key와 value쌍인 것이다.

보통은 input의 경우 key는 LongWritable이고 value는 Text로 고정되는 경향이 있다.

그 이유는 key의 경우에, 원래 input에는 key값이 있을리가 없다.

그런데 값이 들어올때 하둡은 개행 단위로 끊어서 값을 받아오고 그 순서대로 번호를 매긴다.

그리고 그 번호의 값이 key가 된다. 따라서 key값은 숫자가 되고 빅데이터 처리에서 데이터량이 많을 수 있으므로

LongWritable이되는 것이다. 받는 값을 Text로 하는 이유는 그래야 가공이 간편하기 때문이다.

따라서 input의 경우 key는 LongWritable이되고 value는 Text가 된다.

반면에 output의 경우는 여러분이 들어온 값으로 선별한 값이 쓰이게되므로 key와 value는 여러분이 마음대로 정할 수 있다.

위의 솔루션에서는 StringTokenizer로 공백단위로 들어온 텍스트를 끊어내고 그 값을 outputkey로 정하고 그 value는 1로 고정해서 출력된다.

여기서 리듀스로 key,value쌍을 보내는 작업은 context.write이다.

이제 어떻게 동작하는지 이해가 되는가? 들어온 텍스트는 <들어온순서,라인단위 텍스트>로 input값이 들어온다.

반대로 output을 할때는 <끊은 단어,1>값을 output으로 보내게 된다.

리듀스

package hadoop;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();

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

리듀스 작업은 어찌보면 맵과 같다고 할 수있다. 사실 로직은 동일하다.

맵에서 선별한 데이터들을 가지고 작업을 하게 되는것이다.

리듀스도 총 4개의 인자를 가지는데 맵과 같다 input key,input value,output key,output value이다.

당연히 맵의 output인자와 리듀스의 input인자는 같아야한다.

이 솔루션에서 한가지 특이한점은 reduce에서의 value는 map과는 달리 복수형으로 일종의 컬렉션을 받게된다는 것이다.

즉 하나의 값이 아니라 복수의 값을 받게된다. 그 이유는 key가 중복될 수 있기 때문이다.

리듀스 작업이 맵과 다른점이 있다면 맵은 받는 순서대로 값을 받으므로 input key는 중복될 수 없다.

그러나 리듀스는 맵의 output을 받게되는데 맵의 output은 당연히 리듀스의 input이고

맵의 output의 key는 중복이안된다는 제약은 없다. 따라서 리듀스의 input역시 key는 중복될 수 있다.

리듀스의 실질적은 역활은 중복된 키들을 어떻게 처리하느냐이다.

맵은 의미가 같다고 생각되는 데이터를 key를 묶어서 보낸다. 즉 key를 동일하게 보내는 것이다.

그러면 리듀스에서는 그 동일한 key를 묶어서 처리를 한다고 보면된다.

위의 솔루션에서는 value들은 모두 1로 고정되어있다.

그런데 같은 key의 경우에 sum을 계속해서 더하게 된다는 것이다.

예를들어서 들어가는 단어에서 hello라는 단어가 10개가 들어왔다고 가정해보자.

그러면 <hello,1>이라는 output이 맵에의해 탄생했을것이고 리듀스에서는 저 hello라는 key가 10개가 있으므로

10번 for문을 돌면서 sum의값을 증가시키게 될 것이라는 것이다.

드라이버

package hadoop;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class WordCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        if (args.length != 2) {
            System.out.println("Usage: WordCount <input> <output>");
            System.exit(2);
            ;
        }
        Job job = new Job(conf, "WordCount");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);
    }
}