[25일차] DelayCountWithMultipleOutputs.java

---

🤍 2022-07-25 🤍

 

 

DelayCountMapperWithMultipleOutputs.java

 

package com.gyuone.mapper;

import java.io.IOException;

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

import com.gyuone.common.AirlinePerformanceParser;
import com.gyuone.common.DelayCounters;

public class DelayCountMapperWithMultipleOutputs extends Mapper<LongWritable, Text, Text, IntWritable> {
	private final static IntWritable outputValue = new IntWritable(1);
	private Text outputKey = new Text();

	@Override // map override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		AirlinePerformanceParser parser = new AirlinePerformanceParser(value);

		if (parser.isDepartureDelayAvailable()) {
			if (parser.getDepartureDelayTime() > 0) {
				outputKey.set("D, " + parser.getYear() + "," + parser.getMonth());
				context.write(outputKey, outputValue);
			} else if (parser.getDepartureDelayTime() == 0) {
				context.getCounter(DelayCounters.scheduled_departure).increment(1);
			} else if (parser.getDepartureDelayTime() < 0) {
				context.getCounter(DelayCounters.early_departure).increment(1);
			}
		} else {
			context.getCounter(DelayCounters.not_available_departure).increment(1);
		}

		if (parser.isArriveDelayAvailable()) {
			if (parser.getArriveDelayTime() > 0) {
				outputKey.set("A, " + parser.getYear() + "," + parser.getMonth());
				context.write(outputKey, outputValue);
			} else if (parser.getArriveDelayTime() == 0) {
				context.getCounter(DelayCounters.scheduled_arrival).increment(1);
			} else if (parser.getArriveDelayTime() < 0) {
				context.getCounter(DelayCounters.early_arrival).increment(1);
			}
		} else {
			context.getCounter(DelayCounters.not_available_arrival).increment(1);
		}

	}
}

 

---

DelayCountReducerWithMultipleOutputs.java 

 

package com.gyuone.reducer;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class DelayCountReducerWithMultipleOutputs extends Reducer<Text, IntWritable, Text, IntWritable> {
	private Text outputKey = new Text();
	private IntWritable result = new IntWritable();
	private MultipleOutputs<Text, IntWritable> mos;
	
	@Override
	protected void setup(Reducer<Text, IntWritable, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		mos = new MultipleOutputs<Text, IntWritable>(context);
	}

	@Override
	protected void cleanup(Reducer<Text, IntWritable, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		mos.close();
	}

	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		String[] columns = key.toString().split(",");
		outputKey.set(columns[1] + "," + columns[2]); // month
		int sum = 0;
		for(IntWritable value: values) {
			sum += value.get();
		}
		result.set(sum);
		if(columns[0].equals("D")) {
			mos.write("departure", outputKey, result); // 출력으로 들어갈 디렉토리명 지정 departure 
		}else {
			mos.write("arrival", outputKey, result);
		}
	}
}

---

driver 

DelayCountWithMultipleOutputs.java

 

package com.gyuone.driver;

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.MultipleOutputs;
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;

import com.gyuone.mapper.DelayCountMapperWithMultipleOutputs;
import com.gyuone.reducer.DelayCountReducerWithMultipleOutputs;

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

	}
	
	@Override
	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub
		String[] otherArgs = new GenericOptionsParser(getConf(), args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.out.println("Usage: DelayCountWithMultipleOutputs <in> <out>");
			System.exit(2);
		}
		
		Job job = Job.getInstance(getConf(), "DelayCountWithMultipleOutputs");
		
		job.setJarByClass(DelayCountWithMultipleOutputs.class);
		job.setMapperClass(DelayCountMapperWithMultipleOutputs.class);
		job.setReducerClass(DelayCountReducerWithMultipleOutputs.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		
		MultipleOutputs.addNamedOutput(job, "departure", TextOutputFormat.class,
				Text.class, IntWritable.class);
		MultipleOutputs.addNamedOutput(job, "arrival", TextOutputFormat.class,
				Text.class, IntWritable.class);
		
		job.waitForCompletion(true);
		
		return 0;
	}

}

 

 

---

scp .\AirPerformance-0.5 jar ubuntu22:~/work/jar

mapred historyserver start& -> exit 

 

 

yarn jar ./AirPerformance-0.5.jar com.gyuone.driver.DelayCountWithMultipleOutputs air-input delay-count-mos

 

 

hdfs dfs -ls 

hdfs dfs -ls delay-count-mos

part-r-00000가 0이므로 written에서 0이 찍힌다.

 

 

---

● 기존 항공 운항 데이터 분석 결과를 보면 월 순서대로 출력되지 않았음
● 출력 데이터의 키값 자체가 연도와 월이 합쳐진 하나의 문자열로 인식
● 보조정렬(Secondary Sort)은 키의 값들을 그룹핑하고 그룹핑된 레코드에 순서를 부여하는 방식
● 보조정렬 구현순서
	a. 기존 키의 값들을 조합한 복합키(Composite Key)를 정의, 이 때 키의 값 중에서 어떤 키를 그룹핑 키로 사용할 지 결정
	b. 복합키의 레코드를 정렬하기 위한 비교기(Comparator)를 정의
	c. 그룹핑 키를 파티셔닝할 파티셔너(Partitioner)를 정의
그룹핑 키를 정렬하기 위한 비교기(Comparator)를 정의

 

vi 해서 dd :wq -> atom 으로 따로 저장해두기 

 

scp ./* ubuntu22:~/work/air-data/

 

 

"YEAR","MONTH","DAY_OF_MONTH","DAY_OF_WEEK","FL_DATE","UNIQUE_CARRIER","TAIL_NUM","FL_NUM","ORIGIN_AIRPORT_ID","ORIGIN","ORIGIN_STATE_ABR","DEST_AIRPORT_ID","DEST","DEST_STATE_ABR","CRS_DEP_TIME","DEP_TIME","DEP_DELAY","DEP_DELAY_NEW","DEP_DEL15","DEP_DELAY_GROUP","TAXI_OUT","WHEELS_OFF","WHEELS_ON","TAXI_IN","CRS_ARR_TIME","ARR_TIME","ARR_DELAY","ARR_DELAY_NEW","ARR_DEL15","ARR_DELAY_GROUP","CANCELLED","CANCELLATION_CODE","DIVERTED","CRS_ELAPSED_TIME","ACTUAL_ELAPSED_TIME","AIR_TIME","FLIGHTS","DISTANCE","DISTANCE_GROUP","CARRIER_DELAY","WEATHER_DELAY","NAS_DELAY","SECURITY_DELAY","LATE_AIRCRAFT_DELAY",​

 

 

기존에 있는 input csv날리기 hdfs dfs -rm air-input/*

put은 새로운 csv저장하기 hdfs dfs -put ./* air-input/

 

 

다른 input으로 집어넣기 sort하는 csv파일로 바꾸기/ ls로 잘 들어갔는지 확인 

 

---

DataKey.java

 

package com.gyuone.common;

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

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableUtils;

public class DateKey implements WritableComparable<DateKey> {
	private String year;
	private Integer month;

	public DateKey() {
	}

	public DateKey(String year, Integer month) {
		super();
		this.year = year;
		this.month = month;
	}

	public String getYear() {
		return year;
	}

	public void setYear(String year) {
		this.year = year;
	}

	public Integer getMonth() {
		return month;
	}

	public void setMonth(Integer month) {
		this.month = month;
	}

	@Override
	public int compareTo(DateKey o) { // 두개 값을 비교할 때 왼쪽이 크면 +1 같으면 0 뒤에꺼가 크면 -1 얘도 마찬가지
		// TODO Auto-generated method stub
		int result = year.compareTo(o.year);
		if (result == 0) { // year가 같으면 if문 안들어감
			result = month.compareTo(o.month); // 추가로 compareTo
		}
		return result;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		// TODO Auto-generated method stub
		WritableUtils.writeString(out, year);
		out.writeInt(month); // int타입으로 write해라
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		// TODO Auto-generated method stub
		year = WritableUtils.readString(in);
		month = in.readInt(); // 하둡 document에 정의되어 있다.
	}

	@Override
	public String toString() {
		// TODO Auto-generated method stub
		return new StringBuilder().append(year).append(",").append(month).toString(); // key를 새로 생성해준다.
	}
}

점심....

---

● 리듀서에는 그룹핑 파티셔너와 그룹핑 comparator에 의해 같은 연도로 그룹핑된 데이터가 전달된 상태
● 복합키 comparator로 인해 그룹핑된 값은 월의 순서대로 오름차순으로 정렬되어 있음.. 정렬된 상태로 들어온다.
● 하지만 리듀서 메서드에서 지연 횟수를 합산할 경우 데이터에 오류가 발생
● 예를 들어, 2008년도 항공 출발 지연 데이터를 처리할 경우 다음과 같은 결과가 나타남
		EX) 2008 12 2647363
● 2008년도 12월만 출력되고 지연 횟수도 2008년도의 모든  지연 횟수가 합산되어 출력됨
● 이러한 현상이 나타나는 이유는 리듀서는 2008년 이라는 그룹키를 기준으로 연산을 수행하기 때문
● 월별로 지연 횟수를 계산하려면 복합키를 구분해서 처리하는 코드를 구현해야 함
● 입력 데이터의 값에 해당하는 Iterable 객체를 순회할 때 월에 해당하는 값을 bMonth 라는 변수에 백업
● 순회를 하면서 백업된 월과 현재 데이터의 월이 일치하지 않을 때는 
  리듀서의 출력 데이터에 백업된 월의 지연 횟수를 출력
● 이 때, 다음 순서에 있는 월의 지연 횟수를 합산할 수 있게 지연 횟수 합계 변수를 0으로 초기화

 

---

DateKeyComparator.java

package com.gyuone.common;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class DateKeyComparator extends WritableComparator {
	protected DateKeyComparator() {
		super(DateKey.class, true);
	}

	@Override
	public int compare(WritableComparable a, WritableComparable b) {
		// TODO Auto-generated method stub
		DateKey k1 = (DateKey)a;
		DateKey k2 = (DateKey)b;
		
		int cmp = k1.getYear().compareTo(k2.getYear());
		if(cmp !=0 ) {
			return cmp;
		}
		return k1.getMonth() == k2.getMonth() ? 0:(k1.getMonth() < k2.getMonth() ? -1: 1);
		// k1 같으면 0 리턴 다르면 ()로 가서 같으면 -1 리턴 다르다 1 리턴 
		// 삼항연산자를 사용
	}
	
	
}

---