맵리듀스 개요

맵리듀스는 HDFS에 저장된 파일을 분산 배치 분석을 할 수 있게 도와주는 프레임워크다.

개발자는 맵리듀스 프로그래밍 모델에 맞게 애플리케이션을 구현하고, 데이터 전송 및 분산 처리, 내고장성 등 복잡한 처리는 맵리듀스 프레임워크가 자동으로 처리해준다.

 

요즘은 맵리듀스도 옛날 프레임워크라고 하지만, 그래도 맵리듀스를 이해하고 넘어가야 새로운 기술도 바로바로 이해할 수 있을 것이다(맵리듀스를 기반으로 신기술이 나온거라서...)

 

1. 맵리듀스의 개념

맵리듀스 프로그래밍 모델은 맵(Map), 리듀스(Reduce) 두 가지 단계로 데이터를 처리한다.

• 맵: 입력 파일을 한 줄씩 읽어 데이터를 변형
• 리듀스: 맵의 결과 데이터를 집계

 

예시로, 맵리듀스 프로그래밍 모델로 입력 파일의 단어 개수를 계산한다고 해보자.

 

맵은 한 줄에 있는 단어 개수를 계산해 한 줄씩 출력한다.

리듀스는 맵의 출력 데이터를 집계한다.

맵의 출력 데이터에는 단어의 개수로 1씩만 출력되어 있었지만 리듀스의 출력 파일ㄹ에서는 a는 2, book은 2로 집계되어 출력된 것이다.

 

그래서, 맵리듀스 프로그래밍 모델은 아래와 같은 함수로 표현된다.

• 맵: (k1, v1) -> list(k2, v2)
• 리듀스: (k2, list(v2) -> (k3, list(v3))

해석하면 다음과 같다.

• 맵은 키(k1)와 값(v1)으로 구성된 데이터를 입력받아 이를 가공하고 분류, 새로운 키(k2)와 값(v2)로 구성된 목록을 출력한다.
• 리듀스는 새로운 키(k2)로 그룹핑된 값의 목록(list(v2))를 입력 데이터로 전달받고, 값의 목록(list(v2))에 대한 집계 연산을 실행해 새로운 키(k3)로 그룹핑된 새로운 값(v3)의 목록을 생성한다.

 

2. 맵리듀스 아키텍처

맵리듀스 아키텍처를 이해하면 정렬이나 병합 등 다양한 분석 코드를 구현할 수 있게 되고, 환경설정 파일을 튜닝할 때도 도움을 받을 수 있다.

시스템 구성

맵리듀스 시스템은 클라이언트, 잡트래커, 태스크트래커로 구성된다. 각 내용은 다음과 같다.

1. 클라이언트: 사용자가 실행한 맵리듀스 프로그램과 하둡에서 제공하는 맵리듀스 API를 의미하며, 잡(job) 실행을 잡트래커에 요청하고, 잡트래커로부터 잡 진행 상황 및 완료 결과를 공유받는다.
2. 잡트래커: 마스터 서버의 역할로, 태스크트래커에 태스크 실행을 요청하고 하트비트를 받으며 각 태스크트래커의 상태를 파악한다.
3. 태스크트래커: 슬레이브 서버의 역할로, 태스크를 실행하고 모니터링하는 역할을 수행한다.

 

데이터 플로우

다음과 같은 단계로 구성된다.

1. 맵 단계: 입력 파일을 읽어 맵의 출력 데이터를 생성하는 맵 처리 단계
2. 셔플 단계: 맵 태스크의 출력 데이터는 '중간 데이터'이며, 리듀스 태스크는 이 데이터를 내려받아 연산을 수행해야 한다. 이 작업이 진행될 수 있도록 하는 것이 '셔플(shuffle)'로 맵 태스크의 출력 데이터가 리듀스 태스크에게 전달되는 일련의 과정이다.
3. 리듀스 단계: 사용자에게 전달할 출력 파일을 생성하는 단계.

 

위의 예시에서, {read 1 a 1 book 1 write 1 a 1 book 1}이 '중간 데이터'라고 이해하면 된다.

 

맵리듀스 프로그래밍 요소

데이터 타입: WritableComparable

• 맵리듀스는 네트워크 통신을 위한 최적화된 객체로 WritableComparable 인터페이스를 제공한다.
• 맵리듀스 프로그램에서 키와 값으로 사용되는 모든 데이터 타입은 반드시 WritableComparable 인터페이스가 구현되어 있어야 하며, 하둡에서 제공하는 기본적인 데이터 타입은 모드 WritableComparable이 구현되어 있다.
• Writable 인터페이스와 Comparable 인터페이스를 다중 상속한 인터페이스다.
• Comparable 인터페이스는 java.lang 패키지의 인터페이스로, 정렬을 처리하기 위해서 copareTo 메서드를 제공한다.

일단 요약은 이렇게 하긴 했는데, 핵심은 다음이다.

 

직접 데이터 타입을 정의하고 싶으면 WritableComparable 인터페이스를 구현하면 된다.

 

맵리듀스 API는 Wrapper 클래스를 제공하는데, 그 클래스 목록은 다음과 같다.

직접 정의하기 귀찮으면 아래 클래스를 가져다 쓰면 된다.

클래스명	대상 데이터 타입
BooleanWritable	Boolean
ByteWritable	단일 Byte
DoubleWritable	Double
FloatWritble	Float
IntWritable	Integer
LongWritable	Long
TextWrapper	UTF8 형식 문자열
NullWritable	데이터 값이 필요 없을 경우에 사용

 

다음 예시를 보면서 어떻게 데이터 타입을 구현하는지 알아보자.

import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class MyWritableComparable implements WirtableComparable {
	
    private int counter; 
    private long timestamp;
    
    public void write(DataOutput out) throws IOException {
    	out.writeInt(counter);
        out.writeLong(timestamp);
   	}
	public void readFields(DataInput in) throws IOException {
    	counter = in.readInt();
        timestamp = in.readLong();
	}
    
    @Override
    public int compareTo(object o) {
    	MyWritableComparable w = (MyWritableComparable)o;
        if(counter > w.counter( {
        	return -1;
        } else if(counter < w.counter) {
        	return 1;
        } else {
        	if(timestamp < w.timestamp) {
            	return 1;
            } else if(timestamp > w.timestamp) {
            	return -1;
            } else {
            	return 0;
            }
        }
    }
}

int 타입(counter)과 long 타입(timestamp) 변수를 동시에 가지고 있는 데이터 타입인 MyWritableComparable을 구현한 예시다. 이런식으로 구현이 된다는 것만 짚고 넘어가자.

---

InputFormat

입력 스플릿을 맵 메서드의 입력 파라미터로 사용할 수 있게 InputFormat이라는 추상 클래스를 제공한다.

다음 예제는 InputFormat 클래스다. 짧으니 한번 짚고 넘어가보자.

public abstract class InputFormat<K, V> {
	public abstract List<InputSplit> getSplits(JobContext context)
    	throws IOException, InterruptedException;
        
    public abstract RecordReader<K, V> createRecordReader(InputSplit split,
    	TaskAttemptContext context) throws IOException, InterruptedException;
}

• getSplits: 입력 스플릿을 맵 메서드가 사용할 수 있도록 하는 메서드
• createRecordReader: 맵 메서드가 입력 스플릿을 키와 목록의 형태로 사용할 수 있게 RecordReader 객체를 생성하는 메서드

맵 메서드는 이러한 RecordReader 객체에 담긴 키와 값을 읽어 분석 로직을 수행한다. 

 

맵리듀스는 다음과 같은 다양한 InputFormat을 제공한다. 참고로 디폴트값은 TextInputFormat이다.

 

InputFormat	기능
TextInputFormat	텍스트 파일 분석시 사용, 개행 문자(\n)을 기준으로 레코드 분류. 키는 라인 번호로 LongWritable 타입을 사용. 값은 라인 내용으로 Text 타입을 사용
KeyValueTextInputFormat	텍스트 파일을 입력 파일로 사용할 때 라인 번호가 아닌 임의의 키 값을 지정해 키와 값의 목록으로 읽게 된다.
NLineInputFormat	맵 태스크가 입력받을 텍스트 파일의 라인 수를 제한할 때 사용
DelegatingInputFormat	여러 개의 서로 다른 입력 포맷을 사용할 경우에, 각 경로에 대한 작업을 위임
CombineFileInputFormat	이 표에 있는 다른 InputFormat들은 파일당 스플릿을 생성하나, 이는 여러 개의 파일을 스플릿으로 묶어 사용한다. 이때, 각 노드와 랙의 위치를 고려해 스플릿을 정함
SequenceFileInputFormat	SequenceFile(바이너리 형태의 키와 값의 목록으로 구성된 텍스트 파일)을 입력 데이터로 쓸 때 사용.
SequenceFileAsBinaryInputFormat	SequenceFile의 키와 값을 임의의 바이너리 객체로 변환해 사용
SequenceFileAsTextInputFormat	Sequence의 키와 값을 Text 객체로 변환해 사용

---

매퍼

맵리듀스 프로그래밍 모델에서 맵 메서드의 기능을 수행, 키와 값으로 구성된 입력 데이터를 전달받아 이 데이터를 가공하고 분류해서 새로운 데이터 목록을 생성한다.

매퍼 클래스를 그대로 사용할 수도 있지만, 대부분 매커 클래스를 상속받아 매퍼 클래스를 새롭게 구현한다.

 

파티셔너

맵 태스크의 출력 데이터가 어떤 리듀스 태스크로 전달될 지 결정한다. 디폴트 값은 HashPartitioner다.

 

맵 태스크의 출력 키와 값, 전체 리듀스 태스크 개수를 파라미터로 받아 'hash % 전체 리듀스 태스크 개수' 형태로 파티션을 계산한다. 이렇게 계산된 결과대로 맵 태스크가 실행된 노드에 파티션이 생성된 후, 맵 태스크의 출력 데이터가 저장된다.

 

리듀서

맵 태스크의 출력 데이터를 입력 데이터로 전달받아 집계 연산을 수행한다.

 

콤바이너 클래스

셔플할 데이터의 크기를 줄이는 데 도움을 주는 클래스로, 매퍼의 출력 데이터를 입력 데이터로 전달받아 연산을 수행한다.

 

OutputFormat

맵 리듀스 잡의 출력 데이터 포맷은, 잡 인터페이스의 setOutputFormatClass 메서드로 설정한 포맷대로 만들어진다.

OutputFormat은 다음과 같이 여러 유형이 있고, 기본값은 TextOutputFormat이다.

 

OutputFormat	기능
TextOutputFormat	텍스트 파일에 레코드를 출력할 때 사용. 레코드를 출력할 때 키와 값의 구분자는 탭을 사용
SequenceFileOutputFormat	시퀀스파일을 출력물로 쓸 때 사용
SequenceFileAsBinaryOutputFormat	바이너리 포맷의 키와 값을 SequenceFile 컨테이너에 쓴다
FilterOutputFormat	OutputFormat 클래스의 래퍼 클래스. OutputFormat 클래스를 편리하게 사용할 수 있는 메서드를 제공
LazyOutputFormat	FileOutputFormat을 상속받은 클래스는 출력할 내용이 없더라도 리듀스의 출력 파일(part-nnnnn)을 생성한다. 이때 LazyOutputFormat을 사용하면 첫 번째 레코드가 해당 파티션(part-nnnnn)으로 보내질 때만 출력 파일을 생성한다.
NullOutputFormat	출력 데이터가 없을 때 사용

---

아까부터 SequceFile에 대한 이야기가 나오는데, 다음과 같이 요약할 수 있다.

• 키와 값의 형태로 데이터를 저장하는 바이너리 파일로, 데이터 저장을 위한 Writer, 데이터 조회를 위한 Reader, 데이터 정렬을 위한 Sorter 클래스 제공. 압축 적용 방식에 따라 세 가지 유형의 Writer를 제공한다.

1. 무압축: 키와 레코드를 원본 그대로 저장. 다만 레코드를 텍스트 파일에 저장하는 것보다 용량을 많이 차지해 거의 사용하지 않음(압축이 없으니)
2. 레코드 압축: 레코드 단위로 데이터를 압축. 블록 단위로 압축하는 것보다는 압축률이 낮은 편
3. 블록 압축: 블록 단위로 데이터를 압축.

 

3. 맵리듀스 실습

wordcount 사용 예시

https://janggeonwu97.tistory.com/48

hadoop 실습 기록

 

wordcount 구현 예시

https://janggeonwu97.tistory.com/49

hadoop 맵리듀스 실습 기록용