맵리듀스 처리 흐름 알아보기 (MapReduce Phases)

맵리듀스는 맵과 리듀스로 구성되어 있습니다.

 

한 단계 더 나아가 실제 데이터를 읽고 쓰는 처리 흐름을 살펴보면, 좀 더 작은 단위로 나뉠 수 있는 것들로 구성되어 있는 것을 알 수 있습니다. 

 

이번 글에서는 아래와 같은 사항들을 중점으로 맵리듀스 처리 흐름을 자세하게 살펴보겠습니다:

 

• 맵리듀스 처리 흐름 개요
• Record Reader
• Map
• Combiner
• Partitioner
• Shuffle and Sort
• Reduce
• Output Format

---

 

맵리듀스 처리 흐름 개요

파일시스템에서 데이터를 읽어서, 가공하고, 다시 파일시스템에 쓰는 맵리듀스 작업은 아래와 같은 처리 흐름을 가집니다:

MR Workflow - Image from [1]

맵리듀스의 job은 분산되어 실행되는 몇 가지 map tasks와 reduce tasks로 나뉩니다. 각 task는 할당된 데이터의 작은 일부분에 대한 처리를 진행하기에 그러한 처리는 클러스터 전체로 나눠서 수행되게 됩니다. map task는 일반적으로 데이터를 load, parse, transform, filter하는 작업입니다. 각 reduce task는 map task output의 일부를 처리하는 역할을 합니다. 중간데이터는 매퍼 task에서 reducer task로 복사되어 그룹되고 집계됩니다. 이러한 간단명료한 패러다임을 통해 단순한 숫자 집계부터 복잡한 조인 연산이나 Cartesian 연산까지 처리가 가능합니다.

 

맵리듀스 잡의 input은 HDFS에 분산되어 저장된 파일들입니다. 하둡에서 이러한 파일들은 하나의 파일을 어떻게 input split들로 나눌지 정의한 input format을 통해 쪼개집니다. 하나의 input split은 map task에 의해 로드되는 덩어리를 바이트 관점으로 바라본 것입니다. 

 

각 map task는 4가지 페이즈(record reader, mapper, combiner, partitioner)로 나뉩니다. map task의 output은 reducer로 보내집니다. reducer task는 4가지 페이즈(shuffle, sort, reducer, output format)로 나뉩니다. map task가 수행되는 노드는 최적화되어 데이터에 가까운 노드가 선정됩니다 [2]. 

 

 

Record Reader (RR)

레코드 리더는 input format에 의해 생성된 input split을 레코드(records)로 변경합니다. 레코드 리더의 목적은 데이터를 레코드로 파싱하는 것이며, 레코드 자체는 파싱하지 않습니다. 레코드 리더는 데이터를 매퍼에게 키/값 형태로 넘깁니다. 보통 이러한 형태에서 키는 위치 정보이고 값은 레코드를 구성하는 데이터 일부입니다. 

 

Map

매퍼 안에서, 사용자가 제공한 코드는 레코드 리더가 넘긴 키/값 쌍에 대해 실행되어 0개 또는 그 이상의 중간결과물인 키/값 쌍을 생성합니다. 어떤 것이 키이고 어떤 것이 값인지는 맵리듀스가 무엇을 위해 실행되는지에 매우 중요한 요소입니다. 키는 데이터가 그룹되는 기준점이고 값은 reducer의 분석과 관련한 정보입니다. 필터, 집계 등 다양한 맵리듀스 디자인 패턴에 따라 이러한 키/값 쌍을 목적에 맞게 선택하여 처리하게 됩니다. 

 

Combiner

combiner는 옵셔널한 로컬라이즈화된 reducer로 map 페이즈에서 데이터를 그룹할 수 있습니다. combiner는 매퍼로부터 중간결과물의 키를 받아 값을 집계하기 위해 사용자가 작성한 메소드를 사용하여 한 매퍼 내의 중간결과물에 대해 수행합니다. 예로, 전체 집계의 수는 부분 집계의 수들을 더한 것과 같기에, 중간에 일부분씩 나누어 집계를 수행하고 그 결과물들을 더해서 최종 결과물을 집계할 수 있습니다. 대부분의 상황에서 이 combiner의 '중간 집계'는 네트워크 상으로 이동해야할 데이터 사이즈를 급격하게 줄여줍니다. (hello, 3)라는 정보는 보내는 것은 (hello, 1)이라는 정보를 3번 보내는 것보다 적은 비용이 들기 때문입니다. 

 

역시 다양한 맵리듀스 패턴에 따라 combiner가 그런 패턴 특성에 맞게 적용될 수 있습니다. 

 

Partitioner

파티셔너는 매퍼로부터 중간결과물인 키/값 쌍을 받아 reducer 당 하나의 샤드가 되도록 샤드로 나눕니다. 디폴트로 파티셔너는 보통 md5sum을 통해 객체의 해시코드를 얻습니다. 이후 파티셔너는 key.hashCode() % (reducer 수)와 같은 계수 처리를 수행합니다. 이것을 통해 키 공간을 랜덤으로 균일하게 reducer 간에 분산할 수 있으나 같은 키를 가진 다른 매퍼에 존재하는 데이터가 같은 reducer에 모일 수 있도록 합니다. 파티셔너의 디폴트 처리방식은 커스텀화 될 수 있으며 정렬과 같은 경우에는 좀 더 진보한 패턴을 사용합니다. 그러나 파티셔너를 변경하는 것은 매우 드문 일입니다. 파티션된 데이터는 각 map task별로 로컬 파일 시스템에 쓰여서 reducer가 pull하기를 기다립니다.

 

Shuffle and Sort

reduce task는 shuffle과 sort 스텝으로 시작합니다. 이러한 step은 모든 파티셔너에 의해 쓰여진 output 파일들을 받아서 reducer가 실행되는 로컬 파일 시스템에 다운로드 받습니다. 이러한 개별 데이터 조각들은 key를 통해 정렬되고 하나의 큰 데이터 리스트를 이룹니다. 이러한 정렬의 목적은 적절한 키를 하나로 합쳐서 reduce task에서 키에 대응된 값들이 쉽게 순회할 수 있도록 하기 위함입니다. 이 페이즈는 커스텀화가 불가능하며 프레임웍이 모든 것을 자동으로 처리합니다. 개발자가 조정할 수 있는 단 한 가지는 어떻게 키가 정렬되고 그룹되는지 Comparator 객체를 명시하는 것입니다. 

 

Reduce

reducer는 그룹된 데이터를 input으로 받아 key 그룹 마다 reduce 함수를 실행합니다. 이 함수는 키와 키와 연결된 모든 값의 iterator를 받습니다. 넓은 범위의 프로세싱이 이 함수에서 수행될 수 있습니다. 데이터는 집계되거나, 필터링되거나, 여러 방식으로 합쳐질 수 있습니다. reduce 함수 수행이 끝나면, 최종적으로 0개 이상의 키/값 쌍을 마지막 단계인 output format에 전달합니다. map 함수와 같이, reduce 함수는 애플리케이션의 코어 로직이기에 잡에 따라서 변경되곤 합니다.

 

Output format

output format은 최종적인 키/값 쌍을 reduce 함수로부터 받아 record writer를 통해 파일로 쓰게 됩니다. 디폴트로, output format은 키와 값을 탭으로 나누고, 레코드들을 newline으로 나눕니다. 이것 역시 커스텀화가 가능하며, 어쨋든 파일은 HDFS에 쓰여지게 됩니다. 레코드 리더와 같이, output format을 커스터마이징하는 것도 가능합니다. 

 

 

Reference

[1] Big Data with Hadoop MapReduce: A Classroom Approach

[2] MapReduce Design Patterns