하둡 1.0 튜토리얼 - (11) 사용자 정의 옵션

 

Configuration 추가

하둡은 사용자가 여러가지 설정을 추가할 수 있도록 해줍니다.

이전에 작성했던 코드 중 Configuration 클래스를 보셨을 것입니다.

사용자 지정 설정은 이 Configuration설정을 통해 가능합니다.

 

지난 포스팅에서 작성했던 코드는 출발 지연 데이터와 도착 지연 데이터가 각각 다른 Mapper와 Driver클레스에 구현되었고,

빌드 설정과 빌드 Artifact또한 따로 생성해 주어야 했습니다.

이번 포스팅에서는 이를 설정으로 해결해주겠습니다.

 

Mapper 수정

먼저 새로운 Mapper클래스를 추가하겠습니다.

DelayCountMapper.java라고 이름 짓고 아래 코드를 작성해줍니다.

public class DelayCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    public String workType;

    private final static IntWritable outputValue = new IntWritable(1);

    private Text outputkey = new Text();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        workType = context.getConfiguration().get("workType");
    }

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        AirlinePerformanceParser parser = new AirlinePerformanceParser(value);
        
        if(workType.equals("departure")){
            if (parser.getDepartureDelayTime() > 0){
                outputkey.set(parser.getYear() + "," + parser.getMonth());
                context.write(outputkey, outputValue);
            }
        } else if (workType.equals("arrival")){
            if(parser.getArrivalDelayTime() > 0){
                outputkey.set(parser.getYear() + "," + parser.getMonth());
                context.write(outputkey, outputValue);
            }
        }
    }
}

 

먼저 setup함수가 추가된 것을 볼 수 있습니다.

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        workType = context.getConfiguration().get("workType");
    }

이것은 map함수가 실행되기 전에 사용자 지정 설정들을 가져올 수 있는 함수입니다.

위의 코드에서는 workType 즉 출발 지연 데이터이냐 도착 지연 데이터이냐의 정보를 가져오는 코드입니다.

그리고 코드 하단의 map함수를 if else문으로 분기하는 것을 볼 수 있습니다.

 

Driver 클래스

Driver클래스에서는 사용자 설정을 가능하게 해주는 코드를 추가하겠습니다.

아래의 코드를 DelayCount.java파일을 만들고 작성해주겠습니다.

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class DelayCount extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration, new DelayCount(), args);
        System.out.println("MR-Job Result: " + res);
    }

    @Override
    public int run(String[] args) throws Exception {
        if (args.length != 2){ // 입출력 데이터 확인
            System.err.println("Usage: DelayCount <input> <output>");
            System.exit(2);
        }

        String[] otherArgs = new GenericOptionsParser(getConf(), args).getRemainingArgs();
        Job job = new Job(getConf(), "DelayCount");


        // 입출력 경로 설정
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 입출력 포맷 설정
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        // Job, Mapper, Reducer 클래스 설정
        job.setJarByClass(DelayCount.class);
        job.setMapperClass(DelayCountMapper.class);
        job.setReducerClass(DelayCountReducer.class);

        // 출력 Key, Value 유형 설정
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.waitForCompletion(true);
        return 0;
    }
}

코드를 보시면 DelayCount는 Configured와 Tool을 상속받고 있습니다.

Configured는 Job에 넣어주고 있는 getConf()를 통해 환경설정 정보를 제어할 수 있게 해주고,

Tool을 통해 사용자 정의 옵션을 조회할 수 있게 합니다. 또한 Tool을 상속받기 위해서는 꼭 run메소드를 구현해야 합니다.

run메소드의 상단에 보면, otherArgs를 GenericOptionsParser의 getRemainingArgs메소드를 통해 받아오는것이 보일겁니다.

이는 GenericOptionParser에 해당하지 않는, (-D나 -fs같은 하둡 설정 제외의), 다른 매개변수들을 받아올 수 있는 기능입니다.

우리의 경우 이전에는 main에서 사용했던 <input>과 <output>에 해당하는 변수를 받아오는 것입니다.

 

Reducer 클래스

Reducer클래스는 지난번과 동일하게 같은 파일을 사용합니다.

 

프로그램 실행

jar 빌드

이번에는 지난 포스팅과 다른 방식으로, 따로 폴더를 만들어주지 않고 처음에 했던 방식으로 jar를 빌드합니다.

 

jar 실행

이제 구현한 프로그램을 실행하도록 하겠습니다.

scp를 사용해 doop01기기에 jar파일을 옮겨주고, 다음 커맨드를 실행합니다.

>  hadoop jar airlinedata.jar com.jyoon.study.airlinedata.DelayCount -D workType=departure input departure_delay_count

위를 실행하면 맵리듀스가 실행됩니다. 역시 4분정도 소요되었습니다.

 

실행이 끝나면 명령을 사용해 결과를 출력해보겠습니다.

> hadoop fs -cat departure_delay_count/part-r-00000 | tail -10

 

다음과 같은 결과를 확인할 수 있습니다.

2008,11	157278
2008,12	263949
2008,2	252765
2008,3	271969
2008,4	220864
2008,5	220614
2008,6	271014
2008,7	253632
2008,8	231349
2008,9	147061

 

동일한 방식으로 도착 지연 집계 결과를 확인하려면 -D workType=arrival로 설정하면 됩니다.

 

참고자료

이 포스팅은 "시작하세요! 하둡 프로그래밍" 책의 예제를 무작정 따라해본 포스팅입니다.

https://wikibook.co.kr/beginning-hadoop-programming-2rev/