MapReduce
고친 사람 github-actions[bot]
MapReduce 는 서버 수백 대에 큰 계산을 나눠 돌려 주는 계산 틀입니다. 개발자는 조각 하나를 처리하는 함수와 결과를 모으는 함수 두 개만 씁니다. 나머지 뒷일은 전부 틀이 맡습니다. 그 대신 계산 단계마다 중간 결과를 디스크에 쓰므로 한 번 돌리는 데 몇 분에서 몇 시간이 걸립니다.
쉽고 빠른 이해
MapReduce 는 큰 데이터를 조각내서 서버 여러 대가 동시에 계산하게 합니다. 몇 년 치 접속 로그에서 페이지마다 방문 수를 세는 일이 그런 계산입니다.
이게 없으면 개발자가 분산 처리를 손수 짜야 합니다. 데이터를 어느 서버에 보낼지, 서버가 죽으면 무엇을 다시 할지를 전부 챙겨야 합니다. 계산 로직보다 그 뒷일이 몇 배 깁니다.
- 입력을 조각으로 자르고 조각마다 맵 함수를 돌려 「이름표와 값」 짝을 뽑습니다
- 같은 이름표끼리 한 서버로 모읍니다
- 모인 값들에 리듀스 함수를 돌려 최종 결과를 냅니다
대가도 있습니다. 단계 사이마다 디스크를 거쳐서 느립니다. 몇 초 안에 답해야 하는 일에는 못 씁니다. 같은 데이터를 여러 번 되풀이해 훑는 계산은 특히 느려집니다.
상세
MapReduce 는 배치 처리용 계산 틀입니다. 배치 처리는 쌓아 둔 데이터를 한꺼번에 처리하는 방식입니다. 밤사이 하루 치 주문을 모아 매출을 집계하는 일이 배치 처리입니다.
이름이 가리키는 두 가지
MapReduce 라는 이름은 두 가지를 함께 가리킵니다. 하나는 계산을 맵과 리듀스 두 함수로 적는 프로그래밍 모델입니다. 다른 하나는 그 모델로 적은 계산을 서버 여러 대에서 실제로 돌리는 실행 틀입니다.
두 가지는 구글의 논문 한 편에서 함께 나왔습니다. 구글은 이 모델과 자기 회사 안에서 쓰던 실행 틀을 같은 이름으로 공개했습니다. 실행 틀 자체의 코드는 공개하지 않았습니다.
그 논문을 보고 오픈 소스로 다시 만든 실행 틀이 Hadoop 의 MapReduce 입니다. 오늘날 개발자가 「MapReduce 로 돌린다」고 하면 대개 Hadoop 의 이것을 말합니다.
모델만 떼어 쓰는 곳도 있습니다. 몇몇 데이터베이스가 이름이 같은 집계 기능을 둡니다. 그런 기능은 맵과 리듀스로 적는 방식만 빌렸을 뿐 서버 수백 대에 나눠 돌리는 실행 틀은 아닙니다.
한 대로는 너무 오래 걸리는 계산
웹 서비스 접속 로그 100테라바이트에서 페이지마다 방문 수를 센다고 해 봅시다. 디스크 하나가 1초에 100메가바이트를 읽으면 다 읽는 데 열흘이 넘게 걸립니다. 서버 1,000대가 나눠 읽으면 20분이 안 걸립니다.
나눠 읽는 것까지는 쉽습니다. 어려운 것은 그 뒷일입니다. 어느 서버가 어느 조각을 맡을지 정해야 합니다. 서버마다 센 값을 한데 모아야 합니다.
서버가 1,000대면 계산 도중에 몇 대는 거의 늘 죽습니다. 죽은 서버가 맡던 조각을 누가 다시 할지도 정해야 합니다. 이런 뒷일을 계산마다 새로 짜면 계산 로직보다 뒷일 코드가 훨씬 길어집니다.
MapReduce 는 이 뒷일을 틀 하나에 모았습니다. 개발자가 틀에 넘기는 것은 두 함수뿐입니다. 나누고 모으고 다시 돌리는 일은 모든 계산이 같은 코드를 씁니다.
키-값 쌍
MapReduce 에서 오가는 데이터는 전부 키-값 쌍입니다. 키-값 쌍은 이름표 하나에 값 하나를 붙인 짝입니다. 「/home 페이지, 방문 1회」가 키 /home 에 값 1 을 붙인 쌍입니다.
키는 무엇끼리 모을지를 정합니다. 같은 키를 단 값은 나중에 한곳으로 모입니다. 그래서 무엇을 키로 삼을지가 계산의 모양을 정합니다.
맵과 리듀스
맵 함수는 입력 한 건을 받아 키-값 쌍을 0개 이상 내놓습니다. 로그 한 줄을 받으면 「그 줄의 페이지 주소, 1」 쌍 하나를 내놓습니다. 맵 함수는 다른 줄을 볼 수 없습니다. 그래서 줄마다 서로 다른 서버에서 동시에 돌려도 결과가 같습니다.
리듀스 함수는 키 하나와 그 키에 붙은 값 목록을 받습니다. /home 과 [1, 1, 1, …] 을 받아 합을 내놓습니다. 키가 다르면 리듀스 함수끼리도 서로를 볼 필요가 없습니다.
두 이름은 함수형 프로그래밍에서 왔습니다. map 은 목록의 원소마다 같은 함수를 적용하는 연산입니다. reduce 는 목록을 하나의 값으로 접는 연산입니다.
단어 세기도 방문 수 세기와 같은 모양의 계산입니다. 맵 함수가 문서 한 줄을 받아 단어마다 쌍 하나를 내놓는다는 점만 다릅니다. 아래 코드와 그림은 이 단어 세기로 봅니다.
단어 세기를 두 함수로 적으면 이렇습니다. 실제 코드가 아니라 모양만 보인 의사 코드입니다. Hadoop 에서 실제로 짤 때는 맵 함수와 리듀스 함수를 각각 Java 클래스로 적습니다.
def map(줄번호, 줄):
for 단어 in 줄.split():
emit(단어, 1) # 단어마다 (단어, 1) 쌍을 내놓는다
def reduce(단어, 횟수들):
emit(단어, sum(횟수들)) # 같은 단어의 1 을 전부 더한다
두 함수 어디에도 서버나 네트워크 이야기가 없습니다. 이 코드가 한 대에서 돌지 천 대에서 돌지는 실행 틀이 정합니다.
잡과 태스크
실행 틀이 계산을 서버에 나눠 주는 단위는 둘입니다. MapReduce 에 맡기는 계산 한 번이 잡(job)입니다. 잡은 여러 개의 태스크(task)로 쪼개집니다. 태스크는 서버 한 대가 맡는 일 한 덩이입니다. 맵 함수를 돌리는 태스크가 맵 태스크, 리듀스 함수를 돌리는 태스크가 리듀스 태스크입니다.
입력은 대개 HDFS(Hadoop Distributed File System)에 있습니다. HDFS 는 큰 파일을 128메가바이트 안팎의 블록으로 잘라 여러 서버에 나눠 담는 분산 파일 시스템입니다.
맵 태스크 하나는 입력 조각 하나를 맡습니다. 조각 하나는 대개 HDFS 블록 하나입니다. 1테라바이트 입력이면 맵 태스크가 8,000개쯤 생깁니다.
리듀스 태스크 수는 개발자가 정합니다. 리듀스 태스크 하나가 출력 파일 하나를 씁니다.
잡 하나에는 관리 역할을 하는 프로세스 하나가 붙습니다. 이 프로세스가 마스터입니다. 마스터는 태스크를 서버에 나눠 주고 어느 태스크가 끝났는지 기록합니다.
셔플
맵과 리듀스 사이에는 개발자가 안 쓰는 단계가 하나 있습니다. 맵 함수가 내놓은 쌍을 키마다 모아 리듀스 태스크 쪽으로 옮기는 단계입니다. 이 단계가 셔플입니다.
셔플은 세 가지 일을 합니다. 쌍마다 어느 리듀스 태스크로 갈지 정합니다. 옮긴 쌍을 키 순서로 정렬합니다. 같은 키의 값을 한 목록으로 묶습니다.
정렬하는 까닭은 셋째 일을 쉽게 하려는 것입니다. 키 순서로 늘어서면 같은 키의 값이 나란히 붙습니다. 그러면 앞에서부터 차례로 읽기만 해도 키마다 한 목록으로 묶입니다.
어느 리듀스 태스크로 갈지는 보통 키의 해시 값으로 정합니다. 해시 값은 키를 정해진 규칙으로 계산해 얻는 숫자입니다. 같은 키는 언제나 같은 숫자가 됩니다. 이 숫자를 리듀스 태스크 수로 나눈 나머지가 그 번호입니다. 그래서 같은 키는 어느 맵 태스크에서 나왔든 언제나 같은 리듀스 태스크에 닿습니다.
아래 그림은 입력 두 조각으로 단어를 세는 모습입니다. 리듀스 태스크는 둘이라고 놓았습니다. 「배」는 2번으로, 나머지는 1번으로 갑니다.
flowchart TD
subgraph 입력["입력 조각"]
I1["조각 1 · 사과 배 사과"]
I2["조각 2 · 배 포도"]
end
subgraph 맵["맵 태스크"]
M1["(사과,1) (배,1) (사과,1)"]
M2["(배,1) (포도,1)"]
end
subgraph 셔플["셔플 · 키마다 모으고 정렬"]
S1["사과 → 1, 1 · 포도 → 1"]
S2["배 → 1, 1"]
end
subgraph 리듀스["리듀스 태스크"]
R1["리듀스 태스크 1 · 사과 2 · 포도 1"]
R2["리듀스 태스크 2 · 배 2"]
end
I1 --> M1
I2 --> M2
M1 --> S1
M1 --> S2
M2 --> S1
M2 --> S2
S1 --> R1
S2 --> R2
맵 태스크 하나가 두 리듀스 태스크 모두에 쌍을 보냅니다. 셔플은 맵 태스크 전체와 리듀스 태스크 전체가 서로 데이터를 주고받는 단계입니다. 그래서 MapReduce 잡에서 네트워크를 가장 많이 쓰는 단계가 대개 셔플입니다.
컴바이너
단어 세기에서는 맵 태스크 하나가 (사과, 1) 을 수천 번 내놓기도 합니다. 이것을 그대로 셔플로 보내면 1 수천 개가 네트워크를 탑니다.
컴바이너는 맵 태스크가 끝난 서버에서 리듀스를 미리 한 번 돌려 두는 함수입니다. (사과, 1) 수천 개가 (사과, 3,000) 하나로 줄어든 뒤 셔플로 나갑니다.
컴바이너는 어느 순서로 몇 번 묶어도 답이 같은 계산에만 씁니다. 합과 최댓값은 됩니다. 평균은 안 됩니다. 조각마다 낸 평균을 다시 평균 내면 원래 평균과 달라지기 때문입니다.
한 잡이 도는 순서
잡 하나는 아래 순서로 돕니다. 맵 태스크의 출력이 어디에 놓이는지가 뒤의 고장 처리와 이어집니다.
- 마스터가 입력을 조각으로 나누고 조각마다 맵 태스크를 만듭니다
- 맵 태스크는 맡은 조각을 읽어 맵 함수를 돌립니다
- 맵 태스크의 출력은 그 서버의 로컬 디스크에 씁니다. 리듀스 태스크 수만큼 나눠 씁니다
- 리듀스 태스크는 맵 태스크가 끝나는 대로 자기에게 배정된 쌍을 각 서버에서 받아 옵니다
- 배정된 쌍을 다 받으면 키 순서로 정렬하고 리듀스 함수를 돌립니다
- 리듀스 태스크의 출력은 HDFS 에 씁니다
3~5단계가 앞에서 본 셔플입니다. 3단계에서 리듀스 태스크 수만큼 나누는 규칙이 셔플 절의 해시 나머지 규칙입니다.
flowchart TD
subgraph 입력["HDFS · 입력"]
B1["블록 1"]
B2["블록 2"]
end
subgraph 맵서버["맵 태스크가 도는 서버"]
M1["맵 태스크 1"] --> L1["로컬 디스크 · 리듀스 태스크 수만큼 나눈 출력"]
M2["맵 태스크 2"] --> L2["로컬 디스크 · 리듀스 태스크 수만큼 나눈 출력"]
end
subgraph 리듀스서버["리듀스 태스크가 도는 서버"]
R1["리듀스 태스크 1 · 정렬 후 리듀스 함수"]
R2["리듀스 태스크 2 · 정렬 후 리듀스 함수"]
end
subgraph 출력["HDFS · 출력"]
O1["출력 파일 1"]
O2["출력 파일 2"]
end
B1 --> M1
B2 --> M2
L1 --> R1
L1 --> R2
L2 --> R1
L2 --> R2
R1 --> O1
R2 --> O2
맵 출력은 HDFS 가 아니라 로컬 디스크에 둡니다. 맵 출력은 리듀스 태스크가 받아 가면 버리는 중간 결과입니다. 여러 서버에 복사해 두는 HDFS 에 쓰면 금방 버릴 데이터에 복사 비용을 치르게 됩니다.
리듀스 함수는 맵 태스크가 전부 끝난 뒤에야 돌 수 있습니다. 마지막 맵 태스크가 어느 키를 내놓을지 끝나기 전에는 모르기 때문입니다.
고장이 나면
마스터는 서버들에 주기적으로 신호를 보내 살아 있는지 확인합니다. 답이 없는 서버는 죽었다고 보고 그 서버의 태스크를 다른 서버에 다시 맡깁니다.
맵 태스크는 끝난 것도 다시 돌립니다. 맵 출력은 그 서버의 로컬 디스크에만 있었습니다. 서버가 죽으면 리듀스 태스크가 받아 가야 할 출력도 같이 사라집니다.
끝난 리듀스 태스크는 다시 돌리지 않습니다. 리듀스 출력은 이미 HDFS 에 여러 벌 복사돼 있습니다.
다시 돌려도 되는 까닭은 맵과 리듀스가 입력만 보고 답을 내는 함수이기 때문입니다. 같은 조각을 두 번 돌리면 같은 출력이 나옵니다. 맵 함수 안에서 외부 데이터베이스에 쓰는 것처럼 바깥을 건드리면 다시 돌릴 때 그 쓰기가 두 번 일어납니다.
뒤처지는 태스크
잡은 가장 늦은 태스크가 끝나야 끝납니다. 태스크 수천 개 중 몇 개가 디스크가 낡은 서버에 걸려 몇 배 느리게 돌기도 합니다. 이런 태스크 몇 개가 잡 전체를 붙잡습니다.
그래서 잡이 끝나 갈 무렵 아직 도는 태스크를 다른 서버에서 한 벌 더 돌립니다. 둘 중 먼저 끝난 쪽의 결과를 쓰고 다른 쪽은 버립니다. Hadoop 은 이 방식을 투기적 실행(speculative execution)이라고 부릅니다.
디스크에 쓰는 대가
현실의 계산은 대개 맵과 리듀스 한 번으로 안 끝납니다. 로그 거르기, 사용자별 집계, 날짜별 집계만 해도 잡이 셋입니다. 앞 잡의 출력이 다음 잡의 입력이 됩니다.
MapReduce 에서 잡 사이의 데이터는 전부 HDFS 를 거칩니다. 앞 잡은 결과를 HDFS 에 여러 벌 복사해 씁니다. 다음 잡은 그것을 다시 디스크에서 읽습니다. 잡이 열 개면 디스크 쓰기와 읽기도 열 번입니다.
flowchart TD
J1["잡 1 · 로그 거르기"] --> H1[("HDFS · 여러 벌 복사해 씀")]
H1 --> J2["잡 2 · 사용자별 집계"]
J2 --> H2[("HDFS · 여러 벌 복사해 씀")]
H2 --> J3["잡 3 · 날짜별 집계"]
J3 --> H3[("HDFS · 최종 출력")]
같은 데이터를 수십 번 되풀이해 훑는 계산은 이 대가가 가장 큽니다. 머신러닝 학습이나 그래프 계산이 그렇습니다. 반복마다 잡 하나를 새로 띄우고 같은 입력을 디스크에서 다시 읽습니다.
Apache Spark 는 이 약점을 겨냥해 나왔습니다. 단계 사이의 중간 결과를 메모리에 둡니다. 오늘날 새로 짜는 배치 계산은 MapReduce 보다 Spark 로 짜는 일이 많습니다.
MapReduce 로 적기 어려운 계산
맵과 리듀스 두 틀에 모든 계산을 끼워 넣어야 한다는 것도 대가입니다. 조인이 그런 계산입니다.
조인은 두 데이터셋을 키로 잇는 계산입니다. MapReduce 에서는 두 입력을 한 맵 단계에 섞어 읽습니다. 맵 함수가 쌍마다 어느 쪽 입력에서 왔는지 표시를 붙입니다. 리듀스 함수가 그 표시를 보고 같은 키의 두 쪽을 맞춥니다. SQL(Structured Query Language) 한 줄이면 되는 일이 Java 클래스 여러 개가 됩니다.
그래서 MapReduce 위에 더 높은 언어를 얹은 도구들이 나왔습니다. 개발자는 SQL 같은 언어로 씁니다. 도구가 그것을 MapReduce 잡 여러 개로 바꿔 돌립니다. Apache Hive 와 Apache Pig 가 그런 도구입니다.
MapReduce 를 고르는 경우
한 번 쌓아 두고 전부 훑어 집계하는 일에 맞습니다. 하루 치 로그로 통계를 내거나 검색용 역색인을 만드는 일이 그렇습니다. 몇 시간 걸려도 괜찮은 대신 서버 수천 대로 늘려야 하는 일입니다.
몇 초 안에 답해야 하는 조회에는 맞지 않습니다. 잡 하나를 띄우는 데만 수십 초가 걸립니다. 들어오는 대로 처리해야 하는 데이터는 스트림 처리가 맡습니다.
이미 Hadoop 클러스터를 운영하고 있다면 오래된 잡들이 MapReduce 로 남아 있는 경우가 많습니다. 디스크만 거치므로 메모리가 적은 서버에서도 큰 데이터를 끝까지 처리해 낸다는 점도 아직 남은 장점입니다.
맞물림
MapReduce 는 Hadoop 을 이루는 부품 가운데 계산을 맡습니다. 혼자서는 데이터를 담지도, 서버를 나눠 쓰지도 못합니다. 데이터는 HDFS 에 두고 계산할 서버는 자원 관리자에게 빌립니다. 거꾸로 Hive 는 MapReduce 를 불러 SQL 을 돌립니다.
HDFS 에서 읽고 HDFS 에 쓴다
MapReduce 잡의 입력과 출력은 보통 HDFS 의 파일입니다. 잡을 나눌 때 MapReduce 가 HDFS 의 관리 서버인 네임노드(NameNode)를 부릅니다. 입력 파일이 어떤 블록으로 되어 있고 블록마다 어느 서버에 있는지 묻습니다.
MapReduce 는 그 답을 보고 블록을 가진 서버에서 맵 태스크를 띄우려 합니다. 데이터를 네트워크로 옮기는 대신 계산을 데이터가 있는 서버로 보내는 것입니다. 이것이 데이터 지역성입니다.
이 조합에서는 저장과 계산이 한 서버에 묶입니다. 지역성을 얻으려면 HDFS 의 저장 서버와 계산 서버를 같은 서버에 띄워야 합니다. 디스크만 더 필요해도 계산 서버를 함께 늘리게 됩니다.
YARN 에게 서버를 빌린다
YARN(Yet Another Resource Negotiator)은 Hadoop 클러스터의 CPU(Central Processing Unit, 중앙 처리 장치)와 메모리를 여러 계산에 나눠 주는 관리자입니다. MapReduce 는 YARN 에게 태스크를 돌릴 CPU 와 메모리를 빌려 씁니다. 아래에서는 이 둘을 묶어 자원이라고 부릅니다.
YARN 의 중앙 관리자는 리소스 매니저(ResourceManager)입니다. 잡을 내는 프로그램인 클라이언트는 잡을 리소스 매니저에게 냅니다.
잡이 들어오면 YARN 이 먼저 그 잡의 마스터를 띄웁니다. YARN 에서는 이 마스터를 애플리케이션 마스터(ApplicationMaster)라고 합니다. 애플리케이션 마스터는 태스크마다 필요한 자원을 리소스 매니저에게 청합니다.
sequenceDiagram
participant 클라이언트
participant 리소스 매니저
participant 애플리케이션 마스터
participant 네임노드
클라이언트->>리소스 매니저: 잡을 낸다
리소스 매니저->>애플리케이션 마스터: 잡의 마스터를 띄운다
애플리케이션 마스터->>네임노드: 입력 블록이 어느 서버에 있나
네임노드-->>애플리케이션 마스터: 블록마다 서버 목록
애플리케이션 마스터->>리소스 매니저: 그 서버들에서 태스크 돌릴 자원을 달라
리소스 매니저-->>애플리케이션 마스터: 자원을 내준다
Note over 애플리케이션 마스터: 받은 자원에 맵 태스크를 띄운다
처음 Hadoop 에서는 MapReduce 가 서버 관리까지 혼자 맡았습니다. 잡 추적기(JobTracker) 하나가 클러스터 전체의 자원과 모든 잡의 태스크를 함께 챙겼습니다. 클러스터가 커지자 그 한 프로세스가 병목이 됐습니다. 그래서 자원 관리 일이 YARN 으로 갈라져 나왔습니다.
갈라진 덕에 한 클러스터에서 MapReduce 와 Spark 가 같은 서버를 나눠 쓸 수 있습니다. 그 대신 부품이 하나 늘었습니다. 잡이 느리거나 안 뜨면 YARN 의 자원 배분과 MapReduce 의 태스크를 둘 다 살펴야 합니다.
Hive 가 MapReduce 잡을 만든다
Hive 는 HDFS 의 파일을 SQL 로 조회하게 해 주는 도구입니다. 이번에는 방향이 반대입니다. Hive 가 MapReduce 를 부릅니다.
Hive 는 SQL 문을 받으면 그것을 MapReduce 잡 여러 개로 바꿉니다. GROUP BY 는 셔플 한 번이 됩니다. 조인은 앞에서 본 섞어 읽기가 됩니다. 개발자는 맵과 리듀스 함수를 안 쓰고도 클러스터 전체에서 집계를 돌립니다.
그 대신 Hive 는 MapReduce 가 잡을 띄우는 시간까지 물려받습니다. 간단한 조회도 잡을 띄우는 데 수십 초가 걸립니다. 그래서 Hive 는 뒤에 MapReduce 대신 다른 실행 엔진을 고를 수 있게 바뀌었습니다.
관련 항목
MapReduce 가 속하는 상위 분류
Hadoop · 배치 처리 · 분산 시스템 · 병렬 처리 · 오픈 소스 · Apache Software Foundation
MapReduce 를 이루는 단계와 부품
맵 · 리듀스 · 셔플 · 컴바이너 · 파티셔너 · 키-값 쌍 · 정렬 · 해시 함수
MapReduce 가 서 있는 Hadoop 부품
HDFS · YARN · 네임노드 · 리소스 매니저 · 애플리케이션 마스터 · 잡 추적기
MapReduce 잡을 만들어 내는 도구
Apache Hive · Apache Pig · SQL · Java
MapReduce 와 같은 일을 두고 겨루는 계산 엔진
Apache Spark · Apache Flink · Apache Beam · Apache Tez · Dataflow 모델
MapReduce 가 고장과 뒤처지는 태스크를 견디는 방식
투기적 실행 · 재실행 · 데이터 지역성 · 하트비트 · 멱등성 · 복제
MapReduce 모델이 빌려 온 생각
함수형 프로그래밍 · map 함수 · 고차 함수 · 분할 정복 · fork-join
MapReduce 로 흔히 푸는 계산
단어 세기 · 역색인 · 조인 · 로그 · 머신러닝 · 외부 정렬
MapReduce 가 안 맞는 처리 방식
스트림 처리 · 대화형 쿼리 · 인메모리 컴퓨팅 · 데이터플로우 그래프
다른 이름: 맵리듀스 · Hadoop MapReduce