MapReduce: Simplified Data Processing on Large Clusters (2004)

MapReduce도 알아봐야겠지

📝 상세 정리

MapReduce

MapReduce는 대규모 데이터셋을 처리하고 생성하기 위한 프로그래밍 모델이자 구현체

2000년대 초반 구글은 크롤링된 데이터, 웹 요청 로그 등 방대한 양의 raw 데이터들을 처리하여 Inverted Index, 그래프 구조, 크롤링도니 페이지 수와 요약 등 많은 프로그램들을 구현했다. 하지만 이런 계산들이 대부분 개념적으로는 간단한 반면 입력 데이터의 양이 매우 방대해서 수백, 수천대의 머신에 계산을 분산시키고자 하였다. 계산을 병렬화하고 데이터를 분산시키고, 장애를 처리하는 등의 과정들의 문제가 붙으면서 단순한 계산은 복잡한 코드가 될 가능성이 높았고, 이에 따라 Google은 이런 세부사항들을 라이브러리 내에 숨기는 새로운 추상화를 설계했다.

GFS 에서 생각했던 것들과 같이 결함 허용, 데이터 분산 및 부하 분산 등과 같은 개념들을 사용해서 분산 컴퓨팅을 구현해낸 것이고, 아래에서 자세히 알아보자.

아키텍쳐

Map, Reduce

MapReduce는 크게 아래 두가지 함수로 이루어진다.

  • {k, v}쌍을 처리하여 중간 {k, v} 쌍을 만드는 map 함수
  • 모든 중간값들을 병합하는 reduce함수

예시로 들자면 다음과 같다.

def map(key: str, value: str): # key: document name, value: document content
	for w in value: # w: words
		EmitIntermediate(w, "1")
	
def reduce(key: str, values: Iter): # key: a word, value: a list of counts
	result: int = 0
	for v in values:
		result += ParseInt(v)
	Emit(AsString(result))

간단하게 본문들에서 단어의 발생 횟수를 출력하는 예시이다.

Grep, URL 접속 빈도, 웹링크 그래프 등도 위와 같이 간단하게 추상화 할 수 있을 것이다.

아키텍쳐

Image Description

아키텍쳐의 모습과 실제로 어떤 순서로 구현되는지 알아보자.

  1. 사용자는 Map작업의 개수 M, Reduce 작업의 개수 R, 그리고 Worker Machine의 개수를 선택한다.
    • 이때 흔히 M = 200,000 R = 5,000, 그리고 워커 머신은 2,000대를 선택한다.
  2. User Program의 MapReduce 라이브러리는 먼저 Input files를 조각당 16~64MB 크기의 M개 조각으로 분할한다. 그 후 프로그램을 모든 워커에 복제하여 실행한다.
  3. 프로그램 복제본중 하나는 Master가 된다. 나머지는 작업을 할당받을 worker이 된다. 마스터는 앞으로 M개의 map 작업과 R개의 reduce 작업을 worker에 할당할 것이다.
  4. map 작업을 할당받은 worker은 split의 내용을 읽고 계산하여 중간 값을 메모리에 저장한다.
  5. 주기정으로 메모리에 저장된 값은 로컬 디스크에 기록되고, 이 결과는 파티셔닝 함수에 의해 R개의 영역으로 분할된다. 이 로컬에 저장된 값의 위치는 Master에게 전달되고, Master는 이후 reduce worker이 요청할때 이 위치를 전달한다.
  6. reduce worker이 Master에게 이 위치를 전달받으면 reduce worker은 map worker의 로컬 디스크로부터 이 데이터들을 읽어온다. 그리고 모든 데이터를 받아온 후 중간 key를 기준으로 데이터를 정렬한다.
  7. reduce worker은 reduce작업을 수행한다.
  8. 모든 map 작업과 reduce 작업이 끝나면 master은 User Program에게 전달한다.
    • 그 결과는 reduce 작업당 하나씩 R개의 출력파일로 저장된다.
    • 다른 MapReduce 호출의 입력으로 전달할수도 있고, 다른 분산 애플리케이션에서 사용할수도 있으니까

또한 더 좋은 성능을 위해 GFS에 저장된 파일들은 Locality를 살려 해당 머신에서 map 작업을 우선 수행할 수 있도록 하고, 파티셔닝 함수와 정렬 보장등의 테크닉을 추가로 적용한다.

결함 허용

분산되는 방식은 알겠는데, 사실 저게 깔끔하게 진행되기만 한다면 아무 문제가 없겠지만 수천대의 컴퓨터를 사용하는 이상 결함이 발생할 가능 성을 없앨 수는 없겠다. 하나하나 알아보자.

worker 장애

Master은 주기적으로 모든 worker에게 ping을 보내고, 일정 시간동안 worker로부터 응답이 없으면 해당 워커는 실패한 것으로 표시한다. 이후 해당 워커가 수행한 모든 map 작업은 초기 idle 상태로 재설정된다.

  • 실패한 머신의 로컬 디스크에 접근할 수 없기 때문!
  • reduce 결과는 다른 곳에 접근 가능하니 재실행하지 않는다.
Master 장애

Master에 체크포인트를 둬서 새로운 복사본을 시작하는 방식으로 구현할 수 있겠지만, Worker과달리 Master은 한 대만 있기 때문에 장애가 발생할 가능성이 낮다. 따라서 MapReduce 계산을 중단한다.

Straggler

머신 하나가 map 혹은 reduce 작업을 수행하는데 모종의 이유로 너무 오래걸린다면, MapReduce 작업이 완료단계에 가까워질때즈음 마스터는 진행중인 나머지 작업들에 대해 백업 실행을 예약한다. 이후 기존 실행이나 백업 실행중 어느 하나라도 완료되면 해당 작업은 완료된 것으로 표시한다.

결론

MapReduce는 위와 같이 간단한 모델과 아이디어로 무거운 작업을 성공적으로 분산시키는데 성공했다. 구글은 이를 이용해 인덱싱 코드가 cpp 3800줄에서 700줄로 줄일 수 있었을 정도였으며, 프로그램 작성은 물론 성능까지 챙기는데에 성공했다.

❔질문 사항

❓ map과 reduce작업의 할당 부분에서 reduce 작업은 map 작업이 모두 끝나야만 가능할텐데 (R개의 파티션으로 나눠지니까), reduce 작업이 어떤 머신에 먼저 할당되어있다면 해당 worker의 CPU는 놀고 있는게 될거고 (Map 작업이 끝나지 않았으니까) 그렇다고 모든 map worker이 끝난 후에 reduce worker이 할당된다면 네트워크 대역폭을 손해보고 있을텐데, 디테일한 구현이 어떻게 되어있을까?

맞는 말이다. 이를 위해 다음과 같이 할당된다고 보면 된다. 초반 -> 2000대 대부분이 map 수행 중반 -> 대부분 머신이 map을 수행하되, 일부 머신이 먼저 reducer이 되어 shuffle 수행 map 완료 직전 -> map straggler 처리, 더 많은 머신이 reducer 수행 모든 map 완료 -> 조기 시작 reducer은 먼저 reduce 수행, 나머지 reducer은 shuffle 작업 시작 이렇게 하면 클러스터 전체 관점에서 어느정도 파이프라이닝이 되면서 훨씬 빠르다.

🔗 참고 자료

  • Dean, J., & Ghemawat, S. (2004). MapReduce: Simplified data processing on large clusters. In Proceedings of the 6th Conference on Symposium on OSDI (Operating Systems Design and Implementation) (Vol. 6, pp. 10-10).