Hadoop - MapReduce 프로그래밍

MapReduce 프로그래밍은 개발자가 직접 해야한다.

 

=자바(Java)

• MapReduce 프로그램을 가장 완벽하게 작성 가능
• 객체 지향 언어
• 많은 개발 경험 필요
• 가장 많은 코드량 발생

= 하이브(Hive)

• SQL로 하둡 데이터를 분석하려는 SNS 엔지니어들의 요구로 SNS 내부에서 개발되었다가 오픈 소스로 공개됨
• 데이터 요약, 쿼리 및 분석을 제공하기 위해 하둡 위에 구축된 데이터웨어 하우스 소프트웨어 프로젝트

= 피그 (Pig)

• 하둡의 복잡한 추출, 변환, 적재 작업을 손쉽게 함
• 다수의 중간 결과물이 필요한 복잡한 분석 쿼리에도 적합
• 대량의 조인 연산이나 중간 테이블이 필요한 경우에 주로 사용
• 관계형 기본 함수 및 사용자 정의 함수로 기능 확장 가능

 

= MapReduce Java 프로그래밍 예제

: Map, Reduce, Driver(Main) 3개의 클래스가 반드시 구현되어야 함.

• MapReduce에서의 인자는 일반 long, int, String 타입을 사용하지 않으며, 대신 LongWritable, IntWritable, Text 라는 하둡 전용 자료형을 사용
  (전송을 가능하게 바이트별로 보내는 솔루션이 내장되어있기 때문)

 

Mapper Class Example

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> {
    // Map 출력키
    private Text outputKey = new Text();
    // Map 출력값
    private final static IntWritable outputValue = new IntWritable(1);

    @Override
    protected void map(LongWritable inputKey, Text inputValue, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
            throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(inputValue.toString());
        while (itr.hasMoreTokens()) {
            // 출력키 설정
            outputKey.set(itr.nextToken());
            // 출력 데이터 생성
            context.write(outputKey, outputValue);
        }
    }
}

• Mapper 클래스를 상속.

 

Reduce Class Example

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);
    }
}

 

 

Job Class Example

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 이름 설정
        job.setJarByClass(WordCount.class); // Job 클래스 설정
        job.setMapperClass(WordCountMapper.class); // Mapper 클래스 설정
        job.setReducerClass(WordCountReducer.class); // Reducer 클래스 설정
        // 입출력 데이터 포맷 설정
        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); // 실행
    }
}

 

 

= 참고링크

• https://kamang-it.tistory.com/entry/Hadoop-03%EB%A7%B5%EB%A6%AC%EB%93%80%EC%8A%A4-%EC%98%88%EC%A0%9CWordCount