[HDFS] MapReduce : simplified Data Processing on Large Clusters

 

* MapReduce: 대용량 데이터를 생성 및 처리를 위한 분산 프로그래밍 모델이자 구현체

 

구글에서 2004년 발표한 다양한 종류의 파생 데이터를 계산하기 위해 탐색된 문서 , 웹 요청 로그와 같은 대량의 원시데이터를 처리하기 위해 만들어진 소프트웨어 프레임워크 이다. 기존의 Hadoop 클러스터에 범용 서버만 추가하여 computing capacity와 storage capacity 및 I/O 대역폭을 확장하였다. 여러 호스트 간의 데이터 및 계산 파티셔닝을 제공하고 데이터 근처에서 병렬로 응용 프로그램 계산을 실행 한다.  흩어져 있는 데이터를 수직화하여, 그 데이터를 각각의 종류 별로 모으고(Map Function), Filtering과 Sorting을 거쳐 데이터를 뽑아내는(Reduce Function) 하는 분산처리 기술과 관련 프레임워크를 의미 한다.

 

MapReduce는 병렬화 , 내결함성, 데이터 배포 및 로드 밸런싱의 복잡한 세부 정보를 숨긴다.

 

 

* Wordcount  process

 

MapReduce는  사용자의 키/값 쌍을 처리하여 일련의 중간 키/값 쌍을 생성하는 맵(Map) 단계와 맵 단계와 동일한 중간 키와 연관된 모든 중간 값들을 병합하는 리듀스(Reduce) 단계로 처리 과정을 나누어 작업을 의미한다.

 

위의 그림은  HDFS 의 MapReduce의 대표적인 예인 wordcount가 되는 과정을 그림으로 표현한 것이다. 일단 단어가
1. input 되면 , 몇개의 숫자만큼 단어들을 2.spliting하고 단어들의 개수를 세며 3.Mapping 한다. 그리고 같은 단어끼리 모아주기 위해 4.Suffling(+sorting) 과정을 하고 최종적으로 단어의 개수를 세서 5.Reducing하여 6.final result 값을 도출해낸다.

 

WordCount는 말 그대로 단어의 개수만 세는 비교적 간단한 작업이지만 , 입력 데이터는 보통 크고 계산은 합리적인 시간 내에 끝내기 위해 수백 또는 수천 개의 기계에 분산되어야 한다.

 

키/ 값 쌍을 분석하고 각 쌍을 사용자 정의 Map 함수에 전달한다. 중간 키 / 값 쌍은 메모리에 버퍼링 된다. 버퍼링 된 쌍은 주기적으로 로컬 디스크에 기록되며 파티셔닝 함수에 의해 n 영역으로 분할 된다. reducer 가 master로 부터 알림을 받으면 원격 프로시저 호출을 사용하여 mapper의 로컬 디스크에서 버퍼링된 데이터를 읽는다.

 

 

* WordCount code

1) Mapper

public static class Map extends Mapper<LongWritable, Text, Text, IntWritable>{
	private final static IntWritable one = new IntWritable(1);
    	private Text word = new Text();
    
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruprtedException{
    	String line = value.toString();
        String Tokenizer tokenizer = new StringTokenizer(line);
        while(tokenizer.hasMoreToken()){		
        	word.set(tokenizer.nextToken());				//Mapper data(input split block = 64mb)를
          	context.write(word,one);					//한 줄씩 읽어온다
        }
    
    }
}

 

2) Reducer

public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable>{
	public IOException, InterruptedException{
    	int sum = 0;
        for (IntWritable val : values){					//Reducer data의 
        	sum += val.get();					//총합(sum)을 구한다.
        }
        context.write(key, new IntWritable(sum));
    }
}

 

* Why use MapReduce?

 

1) Large-Scale Indexing : indexing code is simpler, smaller, and easier to understand.

 

2) 타고난 병행성(유연성) - 병렬처리지원

 

3) 누구든지 임의로 활용할 수 있는 충분한 서버를 이용하여 대규모 데이터 분석 가능

 

4) fault tolerance : MapReduce는 여러 컴퓨터와 서버에 데이터를 분산시켜 운영

 

5) 데이터 배포 및 load balencing

 

6) 대규모 장애에 대한 복원

 

 

* Basic Cluster Components

 

 

작업을 주도하는 master, 작업을 support 하는 slave 로 구성된다.

 

On the master:

 

Namenode (NN) : 데이터 블록이 어디에 DataNode에 있는지 관리

Jobtracker (JT): MapReduce 작업 요청시 작업을 알맞는 TaskTracker에 할당

 

On each of the slave machines:

 

Datanode (DN): 파일 데이터 블록 전송 및 이동

Tasktracker (TT):  software daemon에 의해 제어, Jobtracker로부터 Map, Reduce and Suffle수행한다.
그리고 multiple task slots들을 보관하고 있다

 

* Data Flow of MapReduce

하둡은 input data들을 잘게 쪼깸으로써 데이터 근접성을 최적화 하였다.

하지만 너무 많이 쪼개면 비효율적이다.  그렇기 때문에 적당한 사이즈로 데이터를 분할해야 한다.

Map tasks는 HDFS가 아닌 local disk에 저장된다. 그이유는 overhead가 생기기 때문

Mapper의 개수는 정해줄 필요는 딱히 .. 없지만 reducer들의 숫자는 정해 주어야 한다.

 

Reduce Task는 데이터 인접성의 이점이 없다. reduce Task 하나당 여러개의  Mapper의 출력이다. 네트워크를 통해 reduce task로 Sort map output은 전송된다. 신뢰성 확보를 위 output은 HDFS에 저장된다. reduce task는 input의 size에 의해 통제되지 않지만, 독립적으로 지정된다.

 

* Combiner

Mapper의 데이터를 최소화하여 Reduce task에 보내주는 역할

 

Map -> (Combiner) -> Reducer

 

Conbiner는 필수가 아니다 combiner가 없어도 MapReduce에는 영향을 끼치면 안 된다.

 

 

 

* 직렬화(serialization) : structured object -> byte stream

 

데이터를 네트워크로 전송하기 위해 구조화된 객체를 바이트 스트림으로 전환하는 과정

 

* 역직렬화(deserialization) : byte stream -> structured object

 

데이터를 전달받은 쪽에서 바이트 스트림을 객체로 전환하는 과정

 

->CSV, XML, JSON등의 형식은 알아보기 쉽다는 장점이 있지만, 데이터 구조상 파싱하는 시간이 오래 걸리는 단점이 있음

이를 binary로 변경하면 사람이 읽기에는 어렵지만 파싱 시간이 짧아지기 때문에 성능상 이득을 볼 수 있는데, 데이터가 작은 경우에는 큰 문제가 없을 수 있지만 빅데이터 시스템에서 사용하는 경우 많은 데이터과 큰 용량의 데이터를 처리해야 하기 때문에 문제가 발생할 수 있다.

 

* Hadoop RPC의 장점

1) 간결성(compact) : 간결한 포맷을 사용하여 hadoop에서 중요한 자원인 네트워크 band-width을 절약

 

2) 고속화(fast) : 프로세스간 통신은 분산 시스템을 위한 back-bone을 형성하여 overhead가 작아야하는데 이를 가능하게 함

 

3) 확장성(extensible) : 새로운 요구사항을 만족시키기 위해 변경이 필요하며 protocol의 발전도 직관적이여야 함, 새로운 인자가 추가되어도 모두 잘 작동해야함

 

4) 상호운영성(interoperable) : 다양한 언어로 작성된 클라이언트를 지원