기본적인 흐름은
key + value(Record라 함.) 형태의 데이터를 맵에서 input으로 받아서, 처리 후 중간 과정들을 겪어서 Reduce에 map의 output을 input으로 제공한다.
그럼 각 과정마다 어떻게 작업을 진행하는 건지, 그 과정을 알아보자!
맵 부분
map -1
input(Record) -> Map -> 인메모리 버퍼
인 메모리 버퍼 : mapreduce.task.io.sort.mb 로 크기를 조절하고, default는 100MB이다. 이 버퍼는 이 정해진 용량에서 한계치 default 80프로 가 채워지면, 버퍼에 있는 내용을 디스크에 작성한다.
즉, 매퍼의 결과물을 버퍼(허용치 까지) -> 디스크에 작성 인 것이다.
이 때, 디스크에 쓰기 전에 스레드는 데이터를 리듀스 수에 맞게 파티션을 나눈다!, 그 후 인메모리 정렬과 컴바이너 함수가 있다면 컴바이너 함수가 실행된다.
참고 사항:
컴바이너 함수(Combiner)란?
쉽게 말하면 key-value 형태의 값들이 key 별로 묶여 있지 않는 상태이므로, 리듀스가 작업하기 편하게, 데이터 양을 줄일 수 있게 키 기준으로 value들을 다 묶는 함수! 커스텀이 가능하여 복잡한 경우 커스텀해서 사용
map -2
인 메모리 버퍼 -> 디스크 -> Reduce(이 때는 디스크 및 메모리 혼합)
이 파트에선 디스크에 써진 파일들을 어떻게 관리하고, 이를 어떻게 Reduce에 넘길지를 집고 넘어가자.
일단 디스크에 써진 파일들을 단일 출력으로 병합해야한다. 그 시기는 맵 작업이 끝날 때 병합을 한다. 이 때 정렬 및 컴바이너가 다시 실행된다. 단, 컴바이너가 실행되는 경우는 디스크에 써진 파일이 3개 이상일 때이다.(3개 이상이 아닌 경우 컴바이너가 더 비효율적! 컴바이너 목적에 비적합)
이 과정을 맞춘 단일 파일들은 각 파티션에 맞춰서 리듀스에 전달 된다.(HTTP 통해서)
-> 이 전달하는 워커 스레드 수도 조절가능(mapreduce.shuffle.max.threads, 단 node기준임. map기준이아니다. default는 서버 프로세스의 2배를 의미)
참고 사항 디스크에 쓸 때 파일을 압축해서 쓸 수도 있음. mapreduce.map.output.compress=true
리듀스 부분
세그먼트 -> 라운드 -> 리듀스
맵 파티션들을 어떤 기준을 가지고 받을까?
이 과정을 셔플 - 파티셔너 작업이라 한다. 기본적인 원칙은 같은 키를 가지는 레코드들을 같은 리듀스에서 처리하자이다.
그래서 키값을 해시코드를 받고, 해시 함수를 통해 리듀스에 맵핑한다. 즉, 키 값이 같은 경우 같은 리듀스에서 작업하는 것을 보장한다.
그럼, 각 Mapper에서 이 리듀스에 해당하는 키를 가진 값들을 받은 것을 세그먼트라 한다. 즉, 맵에 개수만큼 형성되고, 이를 리듀스 작업하기 전에 정렬해야한다.(정렬 이유? 리듀스 편하라고..)
이 때, 정렬을 한 번에 하기 힘드니 round라는 개념이 등장한다 그냥 중간 모임이라고 생각하면 된다.(default 10(10개의 세그먼트),mapreduce.task.io.sort.factor)
라운드 단위로 정렬 후 리듀스에 제공하기 전에 라운드 구분없이 전체 정렬 후 제공한다.
리듀스의 출력은 hdfs를 통해 데이터 노드에 저장됨.
지금까지 다른 작업들은 다 mapper가 수행된 로컬 디스크등을 사용한거임~
참고 사항 :
세그먼트가 크기가 작으면 리듀스에 jvm 에 올린다. 즉 힙에 올린다는 말임. 물론, 힙을 이러한 목적으로만 사용하면 안되서 그 비율가 한계치 조절도 가능함.
이 후, 한계치에 도달하면 맵과 같이 디스크에 쓰기를 시작
이러한 과정을 파라미터 튜닝으로 최적화 할 수 있다. 위에 언급한 파라미터는 물론 추가적인 파라미터도 존재, 이는 다른 페이지에서 추가적으로 정리할 예정
이 전체적인 과정을 다른 흐름이지만 쉽게 설명해논 블로그가 있어 첨부해두겠음.
참고 블로그
