MapReduce 의 비용은 맵과 리듀스 사이의 셔플에 몰려 있다

한 줄 요약

MapReduce 는 맵 → (분할·정렬·셔플) → 리듀스 의 틀이고, 사용자가 쓰는 것은 앞뒤 두 함수뿐이다. 가운데의 정렬과 셔플은 프레임워크가 하며 잡 비용의 대부분이 거기서 난다. 컴바이너는 그 가운데를 줄이는 장치이고, 카운터는 그 가운데를 들여다보는 창이다.

왜 이런 모양이 필요했나

수백 대에 흩어진 테라바이트 자료에서 낱말 수를 세야 한다고 하자. 자료를 한 서버로 모으면 네트워크가 먼저 죽는다. HDFS 설계 문서가 내건 원칙이 "계산을 자료 곁으로 옮기는 편이 자료를 옮기는 것보다 싸다" 인 이유다. 그러면 계산을 블록이 놓인 서버마다 나눠 돌려야 하는데, 이번에는 나눠 돌린 결과를 같은 낱말끼리 다시 모아야 한다. 서버가 죽으면 그 몫을 다시 돌려야 하고, 느린 서버는 기다려야 한다.

이 문제를 잡마다 새로 풀면 코드의 대부분이 분산 처리 뒷일이 된다. MapReduce 는 그 뒷일을 프레임워크로 뺐다. [MapReduce 튜토리얼](https://hadoop.apache.org/docs/r3.5.0/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html)은 프레임워크가 태스크 배치와 감시, 실패한 태스크의 재실행을 맡고, 응용은 입출력 위치와 map·reduce 함수만 준다고 정리한다. 대가는 모든 계산을 키와 값의 쌍으로 표현해야 한다는 제약이다. 그 제약 덕분에 "같은 키는 같은 리듀서로 간다" 는 약속 하나로 모으기를 해결한다.

어떻게 동작하나

흐름을 쌍의 모양으로 쓰면 이렇다. 튜토리얼에 실린 그대로다.

(입력) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (출력)

맵 쪽. 튜토리얼에 따르면 맵 수는 보통 입력의 전체 크기, 곧 입력 파일의 블록 수가 좌우한다. 블록 하나가 대개 입력 조각(split) 하나가 되고 조각마다 맵 태스크가 하나 뜬다. 맵이 내놓는 쌍은 곧장 디스크로 가지 않고 메모리 버퍼에 쌓인다. [mapred-default.xml](https://hadoop.apache.org/docs/r3.5.0/hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml)에서 이 버퍼 크기 mapreduce.task.io.sort.mb 의 기본은 100MB 이고(태스크 힙이 180MB 뿐인 실습 이미지는 32MB 로 줄였다), mapreduce.map.sort.spill.percent 기본 0.80 에 닿으면 뒤쪽 스레드가 내용을 정렬해 디스크에 쏟는다(spill). 맵이 끝나면 쏟은 조각들을 하나로 병합한다. 이 파일은 리듀서 수만큼의 구획으로 나뉘어 있고, 각 구획 안은 키로 정렬되어 있다.

어느 키가 어느 구획으로 갈지는 분할기(Partitioner) 가 정한다. 기본은 HashPartitioner 로, 키의 해시를 리듀서 수로 나눈 나머지다. 같은 키는 어느 맵에서 나왔든 같은 번호의 리듀서로 간다. 모으기가 성립하는 근거가 이 한 줄이다.

리듀스 쪽. 튜토리얼은 리듀서가 셔플·정렬·리듀스의 세 단계를 거친다고 설명한다. 셔플에서 모든 맵 출력 가운데 자기 구획만 HTTP 로 받아 오고, 받아 오는 동안 병합 정렬을 해서 같은 키의 값들을 한 줄로 세운다. 그다음 키마다 reduce(키, 값 목록) 을 한 번씩 부른다. 리듀서 수를 0 으로 두면 맵 출력이 그대로 결과가 되고 정렬도 하지 않는다.

리듀서 수 mapreduce.job.reduces 의 기본은 1 이다. 설정하지 않으면 전체 맵 출력이 리듀서 하나로 몰리고 결과 파일도 part-r-00000 하나다. 리듀서를 셋으로 늘리면 결과 파일이 셋이 된다. 리듀서는 키 차례로 불리므로 낱말 세기처럼 받은 키를 그대로 쓰는 잡이라면 각 파일 안은 키 차례다. 다만 튜토리얼이 못 박듯 프레임워크가 리듀서의 출력을 다시 정렬해 주지는 않고, 세 파일을 이어 붙이면 전체 차례는 섞인다. 해시로 나눴기 때문이다.

전체를 한 줄로 정렬하려면 분할기를 바꿔야 한다. [TeraSort 예제](https://hadoop.apache.org/docs/r3.5.0/api/org/apache/hadoop/examples/terasort/package-summary.html)가 그 방법을 보여 준다. 입력에서 키 표본을 뽑아 N-1 개의 경계를 정하고, 경계 구간마다 리듀서를 하나씩 배정한다. 그러면 i 번 리듀서의 출력은 모두 i+1 번보다 작고, 파일을 번호 순서대로 이어 붙이면 전체가 정렬되어 있다. TeraValidate 는 그것을 검사하는 잡이다.

컴바이너와 카운터

낱말 세기의 맵은 (the, 1) 을 수만 번 내놓는다. 이것을 그대로 셔플하면 네트워크로 1 이 수만 번 건너간다. 컴바이너는 맵 쪽에서 미리 (the, 38211) 로 줄여 보내는 지역 집계다. 튜토리얼은 컴바이너가 쏟기 때 돌고, 리듀스 쪽 병합 중에도 돌 수 있으며, 버퍼보다 큰 레코드가 컴바이너를 거칠지는 정해져 있지 않다고 적는다. 정리하면 컴바이너가 몇 번 돌지는 보장되지 않는다. 그래서 한 번 돌든 세 번 돌든 결과가 같은 연산만 컴바이너로 쓸 수 있다. 합과 최댓값은 되고, 평균은 안 된다 — 평균의 평균은 평균이 아니다.

컴바이너가 제 몫을 하는지는 카운터로 본다. 잡이 끝나면 프레임워크가 센 값이 찍힌다. 맵 입력 레코드와 맵 출력 레코드, 컴바이너 입력과 출력 레코드, 리듀스 입력 그룹 수, 쏟은 레코드 수 같은 것이다. 컴바이너 출력이 입력보다 훨씬 작으면 셔플이 그만큼 줄었다는 뜻이다. 낱말 세기라면 리듀스 입력 그룹 수가 곧 결과에 나오는 서로 다른 낱말 수다. 카운터는 잡이 무엇을 했는지 로그를 뒤지지 않고 숫자로 증명하는 가장 싼 길이다.

YARN 위에서 돈다는 것

하둡 2 이후 MapReduce 는 스스로 자원을 관리하지 않는다. [YARN 구조 문서](https://hadoop.apache.org/docs/r3.5.0/hadoop-yarn/hadoop-yarn-site/YARN.html)가 설명하듯 클러스터 전체를 중재하는 ResourceManager 와 앱마다 하나씩 뜨는 ApplicationMaster 로 역할이 나뉘고, MapReduce 잡의 ApplicationMaster 가 MRAppMaster 다. 잡 하나를 내면 컨테이너가 최소 AM 하나, 맵 태스크 수만큼, 리듀서 수만큼 뜬다.

잡이 끝나면 AM 은 이력을 남긴다. 기본 스테이징 디렉터리가 /tmp/hadoop-yarn/staging 이고 중간 완료 디렉터리는 그 아래 history/done_intermediate 다. 잡 이력 서버를 띄우지 않아도 여기에 .jhist(이벤트 이력)와 .summary, 잡 설정 XML 이 사용자별로 쌓인다. 이력 서버는 이것을 옮기고 보여 주는 역할일 뿐, 기록 자체는 AM 이 한다.

현장에서 만나는 모습

첫째, 잡이 클러스터가 아니라 로컬에서 돈다. mapreduce.framework.name 의 기본값은 local 이다. yarn 으로 바꾸지 않은 클라이언트에서 낸 잡은 ResourceManager 에 나타나지 않고 제출한 기계의 JVM 하나에서 조용히 돈다. "왜 ResourceManager 화면에 잡이 없지" 의 가장 흔한 답이다. 실습 이미지는 이 값을 yarn 으로 바꿔 두었다.

둘째, AM 이 들어갈 자리가 없다. yarn.app.mapreduce.am.resource.mb 의 기본은 1536MB 다. NodeManager 가 컨테이너에 내주는 몫이 그보다 작은 클러스터에서는 AM 컨테이너가 들어갈 노드가 없어 잡이 시작조차 하지 못한다. 이 실습 환경처럼 몫이 1024MB 인 곳에서는 AM 크기를 따로 낮춰 두어야 잡이 뜬다. 실습 이미지는 AM 을 384MB, 맵과 리듀스를 256MB 로 잡아 AM 하나와 태스크 둘이 동시에 들어가게 해 두었다.

셋째, 리듀서 하나가 끝나지 않는다. 기본값 1 을 그대로 두었거나, 키 하나에 값이 몰렸다. 앞은 설정으로 고치고, 뒤는 키 설계로 고친다.

넷째, 컴바이너 때문에 숫자가 틀린다. 평균이나 중앙값을 리듀서 코드 그대로 컴바이너로 걸면, 컴바이너가 도는 횟수에 따라 결과가 달라진다. 맞았다가 틀렸다가 하는 버그라 찾기 어렵다.

실무에서 진짜 중요한 것

• 같은 키는 같은 리듀서로 간다. 모으기의 근거는 분할기 한 줄이다.
• 셔플이 비용이다. 맵과 리듀스 사이로 건너가는 바이트를 줄이는 것이 튜닝의 대부분이다.
• 컴바이너는 몇 번 돌지 모른다. 여러 번 적용해도 결과가 같은 연산만 쓴다.
• 리듀서 기본값은 1개다. 결과 파일 수와 병렬도가 모두 이 값에 달려 있다.
• framework.name 기본값은 local 이다. 잡이 ResourceManager 에 없으면 이것부터 본다.
• 카운터로 증명한다. 입력·출력·컴바이너·그룹 수가 잡이 한 일을 숫자로 보여 준다.

다음 실습에서 할 것

YARN 을 켜고 NodeManager 가 등록되었는지 확인한 다음, 책 다섯 편을 HDFS 에 올려 예제 jar 의 낱말 세기 잡을 돌리고 잡 ID 를 적어 둔다. 이력 서버 없이 HDFS 의 done_intermediate 에 남은 잡 이력 파일을 찾아 여는데, 실습 이미지는 이 파일을 사람이 읽는 JSON 줄로 쓰도록 해 두었다. 잡이 끝날 때의 전체 카운터에서 맵 입력·출력과 컴바이너 입력·출력, 리듀스 입력 그룹 수와 출력 레코드를 꺼내 컴바이너가 셔플을 얼마나 줄였는지 계산한다. 리듀서를 셋으로 늘려 결과 파일이 어떻게 나뉘는지 보고, 마지막으로 teragen·terasort·teravalidate 로 전체 정렬을 돌려 검증까지 통과시킨다.