leeyuunsung
commented
1 year ago 10 MapReduce 처리 과정
MapReduce 처리 과정
- InputFile -> InputFormat 을 통해 읽어들임
- InputFormat 이 정의된 InputSplit 단위로 분할됨
- Key-Value 형태로 RecordReader 를 통해 Mapper 의 입력으로 전달됨
- Mapper 에서 사용자가 정의한 작업을 수행
- Mapper 의 처리 결과가 Combiner 로 전달되는데, Combiner 는 Reducer 로 전달되기 전에 미리 Map 의 결과를 처리하는 기능을 지원 (Local Reducer)
- Combiner 사용시 Shuffle 하는 데이터의 양을 줄일 수 있음
- Partitioner 는 같은 Key 를 갖는 데이터는 물리적으로 동일한 Reducer 로 전달되게 함
- Shuffling & Sorting 은 Map 의 결과를 Reducer 로 전달하는 역할을 수행하는데 네트워크 트래픽이 발생함. Key 를 기준으로 Sorting 이 수행됨
- 이후 Reducer 를 거져 OutputFormat 에 맞게 결과를 OutputFile 로 HDFS 에 저장
InputFormat
public interface InputFormat<K, V> {
InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
RecordReader<K, V> getRecordReader(InputSplit split,
JobConf job,
Reporter reporter) throws IOException;
}
- InputFile 로 부터 데이터를 읽는 역할 수행
- 입력 파일이 분할되는 방식(InputSplit), 읽는 방식(RecordReader) 을 정의
InputFormat 종류
- FileInputFormat
- TextInputFormat
- CombineFileInputFormat
- KeyValueTextInputFormat
- SequenceFileInputFormat
- DBInputFormat
- ...
SequenceFile
- Hadoop에서 제공하는 Binary 파일 포맷
- Key-Value pair로 구성
- Binary로 저장되어 있어서 속도가 빠름
- 압축에 따른 포맷
- Uncompressed SequenceFile Format
- Record-compressed SequenceFile Format (Value 만 압축)
- Block-compressed SequenceFile Format (Block 단위 압축)
- 용도
- 작은 파일들을 모아 SequenceFile로 묶어서 사용
- 압축을 통해 shuffling의 트래픽 양을 줄이는데 사용
InputSplit
public interface InputSplit extends Writable {
long getLength() throws IOException;
String[] getLocations() throws IOException;
}- Map 의 입력으로 들어오는 데이터를 분할
- 하나의 Map 에서 처리하는 입력 데이터
- 맵의 입력으로 들어가는 데이터를 분할 하는 방식 정의
RecordReader
- InputSplit 으로부터 데이터를 읽어들이는 역할
- 단순 반복자
Mapper
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
...
public void run(Context context) throws IOException, InterruptedException {
setup(context);
try {
while (context.nextKeyValue()) {
map(context.getCurrentKey(), context.getCurrentValue(), context);
}
} finally {
cleanup(context);
}
}
}- 사용자가 구현한 비즈니스 로직을 처리하는 역할
Combiner
- Map 의 결과는 Shuffle 을 통해 Reduce 에 전달되게 되는데 네트워크를 타고 전달된다
- 따라서 Map 과 Reduce 사이의 데이터 전송을 최소화 하기 위한 장치
- Map 의 결과를 추가적으로 처리하는 역할
- 최대, 최소값 또는 count 에는 Combiner 가 적용되지만 평균값과 같은것엔 적용되지 않는다
Partitioner
public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {
...
public int getPartition(K2 key, V2 value,
int numReduceTasks) {
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
}- Map 의 결과를 어떤 Reduce 에 전달할지 결정하는 역할
- 같은 Key 를 갖는 데이터는 물리적으로 같은 Reducer 로 전달한다
Shuffling and Sorting
- Shuffling은 Map 의 중간 결과 파일을 Reducer 로 전달 (네트워크 트래픽 발생)
- Reducer 작업 실행 전 전달받은 Key 를 기준으로 Sorting 발생
- Sorting 된 결과가 Reducer 로 전달되게 된다
Reducer
public class Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {
...
public void run(Context context) throws IOException, InterruptedException {
setup(context);
try {
while (context.nextKey()) {
reduce(context.getCurrentKey(), context.getValues(), context);
// If a back up store is used, reset it
Iterator<VALUEIN> iter = context.getValues().iterator();
if(iter instanceof ReduceContext.ValueIterator) {
((ReduceContext.ValueIterator<VALUEIN>)iter).resetBackupStore();
}
}
} finally {
cleanup(context);
}
}
}- 사용자가 정의한 Reducer 작업 수행
OutputFormat
- 연산 결과 포맷
OutputFormat 종류
- FileOutputFormat
- TextOutputFormat
- LazyOutputFormat
- SequenceFileOutputFormat
- DBOutputFormat
- ...
09. MapReduce 기초
MapReduce 란?
MapReduce 용어
InputSplit vs Block
MapReduce 란?
MapReduce Word Count 처리 과정 예시