맵리듀스 프로그래밍 소개

 

 

맵리듀스 프로그래밍 소개

맵리듀스 프로그래밍은 기존 프로그래밍과 어떻게 다른가?

 

 

 

 

 

 

 

 

 

맵리듀스 프로그래밍의 특징

 

데이터 셋은 Key, Value의 집합이며 변경 불가(immutable)

 

데이터 조작은 map과 reduce 두 개의 오퍼레이션으로만 가능

이 두 오퍼레이션은 항상 하나의 쌍으로 연속으로 실행됨
이 두 오퍼레이션의 코드를 개발자가 채워야함

 

맵리듀스 시스템이 Map의 결과를 Reduce단으로 모아줌

이 단계를 보통 셔플링이라 부르며 네트웍단을 통한 데이터 이동이 생김

 

 

 

 

 

 

 

 

 

 

 

 

 

 

맵리듀스 프로그래밍의 핵심: 맵과 리듀스

 

Map: (k, v) -> [(k', v')*]

입력은 시스템에 의해 주어지며 입력으로 지정된 HDFS 파일에서 넘어옴
키,밸류 페어를 새로운 키,밸류 페어 리스트로 변환 (transformation)
출력: 입력과 동일한 키, 밸류 페어를 그대로 출력해도 되고 출력이 없어도 됨

 

Reduce: (k’, [v1’, v2’, v3’, v4’, …]) -> (k’’, v'')

입력은 시스템에 의해 주어짐
맵의 출력 중 같은 키를 갖는 키/밸류 페어를 시스템이 묶어서 입력으로 넣어줌
키와 밸류 리스트를 새로운 키,밸류 페어로 변환
SQL의 GROUP BY와 흡사
출력이 HDFS에 저장됨

 

 

 

 

 

 

 

 

MapReduce 프로그램 동작 예시

Word Count

 

key와 value가 둘 모두 필요하므로 key나 value 중 한가지를 랜덤 값으로 임의 배정하는 경우가 정말 많습니다. 위의 그림에서 각 입력 조각들 예를 들어 the brave yellow lion에게 랜덤 키:abc1234를 배정할 수도 있습니다.

 

 

 

 

 

 

 

 

 

MapReduce: 프로그래밍 예제

Word Count Mapper

public static class TokenizerMapper
 extends Mapper<Object, Text, Text, IntWritable>{
 
 private final static IntWritable one = new IntWritable(1);
 private Text word = new Text();
 
 public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
 	StringTokenizer itr = new StringTokenizer(value.toString());
 	while (itr.hasMoreTokens()) {
 	word.set(itr.nextToken());
 	context.write(word, one);
 		}
 	}
 }

 

 

Input: (100, “the brave yellow lion”)

Output: 
[
 (“the”, 1),
 (“brave”, 1),
 (“yellow”, 1),
 (“lion”, 1)
]

 

Map: (k, v) -> [(k', v')*]

Transformation

키,밸류 페어를 새로운 키,밸류 페어 리스트로 변환

 

 

 

 

 

 

 

 

 

 

MapReduce: 프로그래밍 예제

Word Count Reducer

public static class IntSumReducer
 extends Reducer<Text,IntWritable,Text,IntWritable> {
 private IntWritable result = new IntWritable();
 
 public void reduce(Text key, Iterable<IntWritable> values,
 	    Context context
            ) throws IOException, InterruptedException {
 int sum = 0;
 for (IntWritable val : values) {
 sum += val.get();
 	}
 	result.set(sum);
 	context.write(key, result);
 	}
 }

 

Input: ("lion": [1, 1, 1])

Output: ("lion": 3)

 

Reduce: (k’, [v1’, v2’, v3’, v4’, …]) -> (k’’, v'')

SQL의 GROUP BY와 동일
키,밸류 리스트를 새로운 키,밸류 페어로 변환

 

 

 

 

 

 

 

 

 

 

 

 

 

 

MapReduce: Shuffling and Sorting

 

Shuffling

Mapper의 출력을 Reducer로 보내주는 프로세스를 말함
전송되는 데이터의 크기가 크면 네트웍 병목을 초래하고 시간이 오래 걸됨

 

Sorting

모든 Mapper의 출력을 Reducer가 받으면 이를 키별로 소팅

 

 

 

 

 

 

 

 

 

 

 

 

MapReduce: Data Skew

각 태스크가 처리하는 데이터 크기에 불균형이 존재한다면?

병렬처리의 큰 의미가 없음. 가장 느린 태스크가 전체 처리 속도를 결정
특히 Reducer로 오는 데이터 크기는 큰 차이가 있을 수 있음
    - Group By나 Join등에 이에 해당함
    - 처리 방식에 따라 Reducer의 수에 따라 메모리 에러등이 날 수 있음
데이터 엔지니어가 고생하는 이유 중의 하나
    - 빅데이터 시스템에는 이 문제가 모두 존재

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

MapReduce 프로그래밍의 문제점

 

낮은 생산성

프로그래밍 모델이 가진 융통성 부족 (2가지 오퍼레이션만 지원)
튜닝/최적화가 쉽지 않음
- 예) 데이터 분포가 균등하지 않은 경우

 

배치작업 중심

기본적으로 Low Latency가 아니라 Throughput에 초점이 맞춰짐

 

 

 

 

 

 

 

 

 

MapReduce 대안들의 등장

 

더 범용적인 대용량 데이터 처리 프레임웍들의 등장

YARN, Spark

 

SQL의 컴백: Hive, Presto등이 등장

Hive
    - MapReduce위에서 구현됨. Throughput에 초점. 대용량 ETL에 적합
Presto
    - Low latency에서 초점. 메모리를 주로 사용. Adhoc 쿼리에 적합
    - AWS Athena가 Presto 기반

 

요즘은 HIVE, Presto 둘이 조금씩 비슷해지며 경계가 조금씩 희미해져 가는 분위기입니다.