You've got to find what you love.

# MapReduce - 수치요약(최댓값, 최솟값, 총계)

sample_comment.txt (484)

package hadoop.mr.group;

import java.util.HashMap;
import java.util.Map;

public class MRDPUtils {

    // 이 헬퍼 함수는 스택오버플로우 데이터를 맵 자료 구조로 구문 분석한다.
    public static Map<String, String> transformXmlToMap(String xml) {
        Map<String, String> map = new HashMap<String, String>();
        try {

            // 따옴표로 데이터를 분할 할 수 있다는 점을 이용한다.
            String[] tokens = xml.trim().substring(5, xml.trim().length() - 3).split("\"");
            for (int i = 0; i < tokens.length - 1; i += 2) {
                String key = tokens[i].trim();
                String val = tokens[i + 1];
                map.put(key.substring(0, key.length() - 1), val);
            }

        } catch (StringIndexOutOfBoundsException e) {
            System.err.println(xml);
        }
        return map;
    }

    /**
     * # 로그 파일 형태의 로우를 맵 형식(키,값)으로 리턴
     */
    public static Map<String, String> transformLogToMap(String line){
        Map<String, String> map = new HashMap<String, String>();
        String[] menu = new String[]{"seq", "country", "createDate", "userId",
                    "etcCode", "comment", "device", "uniqueCode", "modifyDate"    };

        String[] tokens = line.trim().replaceAll("\"", "").split(","); // 로그 패턴에 따라 변경

        for(int i=0; i<tokens.length; i++){
            map.put(menu[i], tokens[i]);
        }
        return map;
    }
}

package hadoop.mr.group;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class MinMaxCountTuple implements Writable {
    private String min = "";
    private String max = "";
    private Long count = 0L;
    
    public String getMin() {
        return min;
    }

    public void setMin(String min) {
        this.min = min;
    }

    public String getMax() {
        return max;
    }

    public void setMax(String max) {
        this.max = max;
    }

    public long getCount() {
        return count;
    }
    
    public void setCount(long count) {
        this.count = count;
    }
    
    public void readFields(DataInput in) throws IOException {
        // 쓰여진 순서대로 데이터를 읽는다.
        // 유닉스 timestamp에서 신규 Date 객체를 생성한다.
//        min = new String(in);
//        max = new String(in.readLine());
        min = WritableUtils.readString(in);
        max = WritableUtils.readString(in);
        count = in.readLong();
    }
    
    public void write(DataOutput out) throws IOException {
        // 읽을 순서대로 데이터를 쓴다.
        // Date를 나타내기 위해 유닉스 timestamp를 사용한다.
//        out.writeBytes(min);
//        out.writeBytes(max);
        WritableUtils.writeString(out, min);
        WritableUtils.writeString(out, max);
        out.writeLong(count);
    }
    
    public String toString() {
        return min+ "\t" + max + "\t" + count;
    }

    public static class MinMaxCountMapper extends Mapper<Object, Text, Text, MinMaxCountTuple> {
        
        // 출력 키와 값 Writables
        private Text outUserId = new Text();
        private MinMaxCountTuple outTuple = new MinMaxCountTuple();
        
        // 생성일 문자열을 Date 객체로 표맷팅하는 객체
//        private final static SimpleDateFormat frmt = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS");
        
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
            System.out.println("### parsed :"+parsed);
            
            // 최솟값, 최댓값을 찾기 위해 'CreationDate' 필드를 사용
            String strDate = parsed.get("CreationDate");
            System.out.println("### strDate :"+strDate);
            
            // 분류하기 위해 UserId 필드를 사용
            String userId = parsed.get("OwnerUserId");
            System.out.println("### userId :"+userId);
            
            // 문자열을 Date 객체로 파싱
/*            Date creationDate;
            try {
                creationDate = frmt.parse(strDate);
            
                // creationDate로 최솟값, 최댓값 설정
            
            } catch (ParseException e) {
                e.printStackTrace();
            }*/
            
            outTuple.setMin(strDate);
            outTuple.setMax(strDate);
            
            // 코멘트 갯수를 1로 설정
            outTuple.setCount(1);
            
            // 사용자 ID를 출력 키로 설정
            outUserId.set(userId);
            
            // 시간과 평균 코멘트 길이를 씀
            context.write(new Text(outUserId), outTuple);
        }
    }    

    public static class MinMaxCountReducer extends Reducer<Text, MinMaxCountTuple, Text, MinMaxCountTuple> {
        
        // 출력 값 Writable
        private MinMaxCountTuple result = new MinMaxCountTuple();
        
        public void reduce(Text key, Iterable<MinMaxCountTuple> values, Context context) throws IOException, InterruptedException {
            
            // 결과 초기화
            result.setMin(null);
            result.setMax(null);
            result.setCount(0);
            int sum = 0;
            
            // 해당 키에 대한 모든 값을 순환
            for (MinMaxCountTuple val : values) {

                // 입력의 최솟값이 결과의 최솟값보다 작으면, 입력의 최솟값을 결과의 최솟값으로 설정
                if (result.getMin() == null || val.getMin().compareTo(result.getMin()) < 0) {
                    result.setMin(val.getMin());
                }
                
                // 입력의 최댓값이 결과의 최댓값보다 크다면, 입력의 최댓값을 결과의 최댓값으로 설정
                if (result.getMax() == null ||
                        val.getMax().compareTo(result.getMax()) > 0) {
                    result.setMax(val.getMax());
                }
                
                // 값에 대한 개수를 합한다.
                sum += val.getCount();
            }
            
            // 입력 값들의 숫자의 개수를 설정
            result.setCount(sum);
            context.write(key, result);
        }
    }    
}

package hadoop.mr.group;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;

public class MinMaxCountTupleDriver {
    
    public void runJob(String inputPath, String outputPath) throws Exception {

        Configuration conf = new Configuration();
        Job job = new Job(conf, "MinMaxCountTupleDriver driver");
        job.setJarByClass(MinMaxCountTupleDriver.class);
        
/*        job.setPartitionerClass(GroupKeyPartitioner.class);
        job.setGroupingComparatorClass(GroupKeyComparator.class);
        
        job.setSortComparatorClass(DateKeyComparator.class);*/

        // the keys are words (strings)
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(MinMaxCountTuple.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(MinMaxCountTuple.class);

        job.setMapperClass(MinMaxCountTuple.MinMaxCountMapper.class);
        job.setReducerClass(MinMaxCountTuple.MinMaxCountReducer.class);

        FileInputFormat.addInputPath(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
    
    public static void main(String[] args) throws Exception {
        (new MinMaxCountTupleDriver()).runJob("/user/sunshiny/sample_comment.txt"
            , "/user/sunshiny/sample_out");
    }
}