하둡 공부 - MapReduce 작동 방법

 

- MapReduce Job 실행 상세

 

• ResourceManager(RM), NodeManager(NM), Application Master(AM)

 

 

 

• 잡 제출
• 1단계
• Job을 제출하면 waitForCompletion() 메서드가 1초에 한 번씩 변경 내역이 있으면 콘솔로 보여줌
• 2단계
• RM에 새로운 애플리케이션 ID를 요청함
• 3단계
• Job 실행에 필요한 JAR 파일, 환경 설정 파일, 계산된 입력 스플릿 등의 Job 리소스를 공유 파일시스템에 있는 해당 Job ID 이름의 디렉터리에 복사
• 4단계
• RM에게 Job을 제출함
• 잡 초기화
• 5a, 5b 단계
• 스케줄러는 컨테이너를 하나 할당하고, RM은 AM 프로세스를 시작함
• 6단계
• AM을 실행하고, 이후 각 Task로부터 진행 및 종료 보고를 수신
• 7단계
• Client가 계산한 입력 스플릿 정보를 공유 파일시스템에서 읽어옴
• 태스크 할당
• 8단계
• AM은 RM에게 모든 Task 컨테이너를 요청
• Map Task 요청이 Reduce Task 요청보다 우선순위가 높음
• (Slow Start) 전체 Map Task의 5%(기본값)가 완료되기 전까지 Reduce Task의 요청은 처리되지 않음
• 태스크 실행
• 9a, 9b 단계
• RM은 할당받은 컨테이너를 시작함
• 10단계
• Job 환경 설정, JAR 파일, 분산 캐시 등 필요한 리소스를 로컬로 가져옴
• 11단계
• Map과 Reduce Task를 실제 실행
• YarnChild는 별도 JVM에서 실행되어 강제 종료되어도 NM는 영향을 받지 않음

 

- 진행 상황과 상태 갱신

 

• 자식 프로세스는 진행 상황과 상태 정보를 집계하여 3초마다 인터페이스를 통해 AM에게 보고함
• 상태 정보
• Job과 개별 Task의 실행중, 성공적으로 완료됨, 실패와 같은 Job 등의 정보
• 진행 상황
• 처리한 입력 데이터의 비율

 

- 잡 완료

 

• 마지막 Task가 완료되면 AM이 Job의 상태를 성공으로 변경
• waitForCompletion() 메서드가 반환되고, Job 통계와 카운터가 콘솔에 출력됨
• mapreduce.job.endnotification.url 속성을 통해 HTTP Job 통지(콜백)를 설정할 수 있음

 

- 잡 실패

 

• 태스크 실패
• 행(보통 10분(기본값) 동안 진행 상황 갱신을 받지 못함)이나 예외가 발생한 경우 해당 Task를 실패로 간주하고 해당 리소스를 풀어줌
• 실행 시간이 긴 Task의 경우 MapReduce가 진행 중임을 판단하도록 조치(레코드 읽기/쓰기, 카운터 증가 등)가 필요함
• AM은 해당 실패한 Task를 4번(기본값)까지 재시도
• 최대 허용 Task 실패 비율을 설정하여, 일부 Task가 실패해도 Job이 중단되지 않도록 조절할 수 있음
• 노드 매니저 실패
• NM이 충돌에 의해 실패하거나 굉장히 느리면, RM의 하트비트 전송을 중단하거나 드물게 전송
• RM은 NM이 10분(기본값) 동안 하트비트 전송이 중단되면 이를 스케줄링하는 노드 풀에서 제거
• 하나의 NM에서 4개(기본값) 이상의 Task가 실패하면 NM 자체가 실패하지 않더라도 NM은 블랙 리스트에 등록
• 리소스 매니저 실패
• RM의 실패 이벤트 발생시 모든 실행 중인 Job이 실패 (단일 고장점)
• 고가용성(HA)을 달성하기 위해서는 두 개의 RM를 활성-대기 설정으로 실행해야 함
• 실행 중인 애플리케이션에 대한 모든 정보는 고가용 상태 저장소(Zookeeper 혹은 HDFS)에 보관

 

- 셔플과 정렬

 

 

 

• 스플릿 생성
• 분산 처리를 위해 Job의 입력이 분리되는 논리적 고정 크기 조각
• 맵리듀스 옵션을 통해 Max/Min 값을 설정할 수 있음
• 각 Split마다 하나의 Map Task를 생성
• 가장 적절한 Split 크기는 HDFS 블록의 기본 크기인 256MB가 적당함
• Split의 크기가 작을수록 부하분산에 더 좋은 효과
• 너무 작으면 Split 관리와 Map Task 생성을 위한 오버헤드
• 맵
• 메모리 버퍼(기본값 100MB)에 Map Task의 수행 결과를 저장함
• 스필
• 버퍼의 내용이 80%(기본값)에 도달할 때마다 백그라운드 스레드가 전송할 Reducer 수에 맞게 파티션을 나누고 디스크에 스필을 시작함
• 스필이 진행되는 동안에도 Map 결과는 계속해서 버퍼에 쓰임
• 버퍼가 가득차면 Map은 스필이 종료될 때까지 블록됨
• (정렬 및) 병합
• 각 파티션 내 백그라운드 스레드는 키를 기준으로 인메모리 정렬을 수행
• 스필 파일이 3개 이상인 경우 컴바이너로 Reducer에 전송할 데이터양을 줄임
• 압축 옵션은 상황에 따라 디스크에 빨리 쓰고 디스크 공간을 절약하며, Reducer로 전송할 데이터 양을 줄일 수 있음
• 복사
• 5개(기본값) 스레드로 Map 출력의 복사를 병렬로 수행
• AM이 하트비트로 완료 Mapper와 Reducer의 매핑 정보를 생성
• Reducer가 Mapper 출력을 회수한 후에도 오류에 의한 실패가 발생할 수 있기 때문에 Job 종료 후 AM으로부터 삭제 요청을 받을 때까지 기다림
• 정렬(사실상 병합)
• 정렬(병합)은 라운드 단위(Map 출력 개수/병합 계수(기본값 10))로 수행됨
• ex) 40개 -> 4개(4(40/10)개 병합, 10개 병합, 10개 병합, 10개 병합) + 6개 -> Reducer
• Reducer에 전달하는 최종 round에서 개수를 병합 계수로 맞추기 위한 최소한의 병합을 수행하는 것이 목적
• 리듀스
• 최종 라운드는 병합 대신 Reduce 함수에 곧바로 전송하여 디스크 IO를 줄임

 

 

 

참조: Hadoop: The Definitive Guide