맵리듀스 작업 개선을 위한 추가적인 사항들

맵리듀스는 간단하게 map과 reduce 함수 작성을 통해 구현됩니다. 대부분의 경우에 기본적인 map과 reduce만으로 충분하지만, 특정 상황에서는 몇몇의 추가적인 설정이 매우 유용한데요. 이번 글에서는 [1]에 기반해 그러한 사항들을 알아보겠습니다.

---

파티셔닝 함수

맵리듀스 사용자들은 원하는 reduce tasks 수(결과물 파일 수)를 설정하게 됩니다. 중간결과물의 키에 대한 파티셔닝 함수를 사용하여 데이터는 이러한 task들에 파티션되게 됩니다. 디폴트 파티셔닝 함수는 hashing을 사용합니다. 이 경우 대부분 적절히 균형잡힌 파티션이라는 결과는 낳는데요. 하지만 어떠한 상황에서는 데이터를 다른 파티셔닝 함수를 사용해 파티션하는 것이 유용하기도 합니다. 예로, 결과물의 키가 URL이고 단일한 호스트에 대한 모든 진입점이 같은 결과물 파일에 저장되기를 원하는 경우 사용자는 특별한 파티셔닝 함수를 설정할 수 있습니다 (partitioner 참조). "hash(Hostname(urlkey)) mod R"과 같은 파티셔닝 함수는 그러한 목적을 달성할 수 있게 합니다.

 

순서 보장

맵리듀스는 주어진 하나의 파티션 내에서, 중간결과물의 키/값 쌍이 키에 대해 오름차순으로 정렬되는 것을 보장합니다. 이러한 정렬 보장은 파티션별로 정렬된 결과물 파일을 만들게 되는데, 결과물 파일 포맷이 키에 기반한 효율적인 랜덤액세스를 지원하는 것과 같은 경우에 도움이 됩니다. 

 

Combiner 함수

몇몇의 경우에, 각 map task에 의해 생성된 중간결과물 키들에 상당한 반복이 있고 사용자가 작성한 reduce 함수가 commutative하고 associative한 경우가 있습니다. 워드카운트와 같은 사례가 이에 해당되는데요. 단어의 빈도수는 Zipf 분포를 따르고, 각 map task는 수백, 수천개의 <the, 1>와 같은 형태의 레코드를 생성합니다. 이러한 모든 count들은 네트워크를 통해 하나의 reduce task로 보내져 reduce 함수에 의해 합쳐지게 됩니다. 맵리듀스는 사용자가 옵셔널한 Combiner 함수를 설정할 수 있도록 하여 네트워크를 통한 송신 전에 부분적인 합산(partial merging)을 가능하게 합니다. 

 

Combiner 함수는 map task를 실행하는 각 머신에서 실행됩니다. 보통 combiner와 reduce 함수를 구현하기 위해 같은 코드가 사용됩니다. reduce 함수와 combiner 함수의 유일한 차이점은 맵리듀스 라이브러리가 어떻게 결과물을 처리하느냐에 존재합니다. reduce 함수의 결과물은 최종 결과물 파일에 쓰여지나, combiner 함수의 결과물은 중간결과물 파일에 쓰여지고 reduce task에 보내집니다. 

 

이러한 부분 combining은 맵리듀스 연산의 특정 클래스의 속도를 매우 높여줍니다. 

 

Input과 Output 타입

맵리듀스 라이브러리는 다른 포맷의 input 파일 읽기를 지원합니다. 예로, "text" 모드 input은 각 줄을 키/값 쌍으로 처리합니다. 사용자는 간단한 reader 인터페이스를 구현하여 새로운 input 타입을 지원할 수 있습니다. 

 

 reader는 꼭 파일에서만 데이터를 읽는 것이 아니라, 데이터베이스와 같은 원천도 읽을 수 있습니다. 또한, 유사한 인터페이스를 writer로 output에 대해서도 제공합니다. 

 

사이드 이펙트

몇몇의 경우에 맵리듀스는 map과 reduce 오퍼레이터의 보조 파일들을 추가적인 결과물로 생성해주어 특정 목적에 사용합니다. 맵리듀스는 그러한 사이드 이펙트를 atomic하고 멱등하게 만들기 위해서 애플리케이션 writer에 의존합니다. 보통 애플리케이션은 임시 파일에 쓰고  완전히 결과물이 발생되었을 때 파일을 rename합니다. 

 

맵리듀스는 하나의 task에 의해 생성되는 여러 결과물 파일들에 대한 atomic two-phase 커밋을 지원하지 않습니다. 그러므로, 여러 결과물 파일을 만들고 cross-file 일관성 요구사항을 가진 task의 경우는 deterministic해야 합니다. 

 

Bad records 건너뛰기

때때로 사용자 코드에 버그가 있어서 특정 레코드에 map과 reduce 함수가 deterministically 실패하도록 만듭니다. 그러한 버그는 맵리듀스 작업이 완료되는 것을 막는데요. 보통은 이러한 버그를 수정하는 것이 필요하나, 제3자 라이브러리를 사용하여 코드를 살펴볼 수 없어서 수정이 불가능한 경우도 있습니다. 또한, 대용량의 데이터셋의 처리를 하는 경우의 매우 일부의 레코드는 무시해도 괜찮은 경우가 있습니다. 맵리듀스는 옵셔널한 모드를 제공하여 deterministic한 에러를 발생하는 레코드의 경우 무시하고 처리를 계속 수행할 수 있도록 합니다. 

 

로컬 실행

map 또는 reduce 함수의 문제를 디버깅하는 것은 매우 복잡합니다. 이는 실제 컴퓨팅이 수천개의 머신들로 이뤄지고 할당이 마스터에 의해 자동적으로 수행되는 분산 시스템에서 수행되기 때문인데요. 디버깅, 프로파일링과 작은 스케일의 테스트를 지원하기 위해서, 맵리듀슨는 맵리듀스의 모든 연산을 로컬 머신에서 순차적으로 수행하는 추가적인 라이브러리를 구현해두었습니다. 사용자는 컴퓨팅이 제한된 특정 map task들에만 적용되도록 설정할 수 있습니다. 사용자는 특별한 플래그와 함께 프로그램을 실행하여 디버깅 또는 테스팅 도구를 사용할 수 있습니다 (gdb 등).

 

상태 정보

마스터는 내부 http 서버를 실행하여 사용자에게 상태 페이지를 제공합니다. 상태 페이지는 몇 개의 task가 종료되었는지, input의 크기, 중간결과물의 크기, 최종 결과물의 크기, 프로세싱 비율 등의 컴퓨팅의 진척을 보여줍니다. 또한, 그러한 페이지들은 각 task에서 발생한 stderr, stdout 아웃풋에 대한 링크도 포함하고 있습니다. 사용자는 이러한 데이터를 사용하여 얼마나 컴퓨팅이 걸릴지, 컴퓨팅에 더 많은 리소스가 필요하지는 않은지와 같은 부분을 살펴볼 수 있습니다. 또한, 컴퓨팅이 예상보다 느릴 때 이러한 페이지에서 제공하는 정보를 사용할 수 있습니다. 

 

추가적으로, 탑레벨의 상태페이지는 어떤 워커가 실패하였는지, 어떤 map과 reduce task들이 실패 당시 어떠한 task를 실행했는지와 같은 사항들도 알려줍니다. 

 

카운터

맵리듀스 라이브러리는 카운터 기능을 통해 다양한 이벤트 발생 빈도를 카운팅할 수 있도록 합니다. 예로, 사용자 코드는 처리된 모든 단어의 갯수를 세거나, 인덱스된 독일어 문서 갯수를 세는 것 등을 원할 수 있습니다. 

 

이러한 기능을 사용하기 위해서, 사용자 코드는 named 카운터 객체를 생성하고 map 및 reduce 함수 안에서 적절하게 카운터를 증가시킵니다. 

 

개별 워커 머신의 카운터 값은 주기적으로 마스터에 전파됩니다. 마스터는 성공적인 map과 reduce task들로부터의 카운터 값을 집계하고 맵리듀스 작업이 끝나면 그러한 값을 사용자 코드에 전달합니다. 현재 카운터 값들은 또한 마스터 상태 페이지에 노출되어 사용자가 live 컴퓨팅을 확인할 수 있도록 합니다.

 

Reference

[1] MapReduce: Simplified Data Processing on Large Clusters