Mapreduce-Job-Process

맵리듀스 잡은 클라이언트가 잡 실행을 요청하는 단계, 해당 잡이 초기화되는 단계, 잡을 실행하기 위한 태스크를 할당하는 단계, 할당된 태스크가 실행되는 단계, 마지막으로 잡이 완료되는 단계로 구성됩니다.

1. 잡 실행 요청

맵리듀스 애플리케이션인 클라이언트가 잡 실행을 요청하면 아래와 같이 작업이 진행됩니다.

1. 클라이언트는 org.apache.hadoop.mapreduce.Job의 waitForCompletion 메서드를 호출해 잡 실행을 요청합니다. 이때 클라이언트의 요청은 Job의 내부 컴포넌트인 JobClient에 전달됩니다.

2. JobClient는 잡트래커의 getNewJobId 메서트를 호출해 새로운 잡ID를 요청합니다. 잡트래커는 잡의 출력 파일 경로가 정상적인지 확인한 후 잡ID를 발급합니다. 참고로 JobClient와 잡트래커는 RPC로 통신하며, JobSubmissionProtocol에 정의된 프로토콜을 사용합니다.

3. 클라이언트는 잡을 실행하는 데 필요한 정보를 잡트래커와 태스크트래커에게 공유해야 합니다. 그래서 JobClient는 다음과 같은 파일을 HDFS에 저장합니다.

   • 입력 스플릿 정보
   • JobConf에 설명된 정보
   • 잡 클래스 파일 혹은 잡 클래스가 포함된 JAR 파일

4. JobClient는 잡트래커의 submitJob 메서드를 호출해 잡 실행을 요청합니다.

2. 잡 초기화

잡트래커는 잡을 실행하기 위한 초기 설정 작업을 진행합니다. 각 과정은 아래와 같이 진행됩니다.

1. 잡트래커는 잡의 상태와 진행 과정을 모니터링할 수 있는 JobInProgress를 생성합니다. JobInProgress는 JobClient가 HDFS에 등록한 잡 공통 파일을 로컬 디스크로 복사한 후, 스플릿 정보를 이용해 맵 태스크 개수와 리듀스 태스크 개수를 계산합니다. 또한 잡의 실행 상태를 RUNNING으로 설정합니다.
2. 잡트래커는 1단계에서 생성한 JobInProgress 객체를 내부 큐인 jobs에 등록합니다. 큐에 등록된 JobInProgress는 스케줄러에 의해 소비됩니다.

3. 태스크 할당

맵리듀스는 태스크를 할당하기 위한 스케줄러인 TaskScheduler를 제공합니다. TaskScheduler는 추상 클래스이며, 이를 구현한 FIFO 방식의 JobQueueTaskScheduler, 풀 방식의 FairScheduler, 다중 큐 방식의 CapacityScheduler를 제공합니다. 기본 스케줄러는 FIFO 방식의 스케줄러를 이용해 잡의 실행 순서대로 태스크가 할당됩니다.

과정은 아래와 같이 진행됩니다.

1. 태스크트래커는 3초에 한 번씩 잡트래커에게 하트비트 메시지를 전송하며, 이를 통해 태스크트래커가 실행 중이라는 것과 새로운 태스크를 실행할 준비가 됐다는 것을 알려줍니다.

2. 스케줄러는 태스크트래커의 하트비트 메시지를 확인한 후, 내부 큐에서 태스크를 할당할 잡을 선택합니다. 그리고 해당 잡에서 하나의 태스크를 선택합니다. 이때 잡의 선택은 각 스케줄러의 알고리즘에 맞게 선택하게 됩니다.

3. 스케줄러는 맵 태스크와 리듀스 태스크를 구분해 태스크를 할당합니다. 우선 맵 태스크의 경우 입력 스플릿과 동일한 서버의 태스크를 선택합니다. 이는 네트워크를 통하지 않고, 로컬 디스크에 접근해서 높은 성능을 낼 수 있기 때문입니다. 차선으로는 동일한 랙의 태스크를 선택합니다. 리듀스 태스크의 경우는 대부분 맵 태스크의 출력 데이터를 네트워크로 내러받기 때문에 단순히 태스크 목록에 있는 순서대로 선택하게 됩니다.

스케줄러는 태스크를 선택한 후 해당 태스크트래커에게 태스크 할당을 알려줍니다. 잡트래커는 태스크트래커가 전송한 하트비트의 응답으로 HeartbeatResponse를 전송합니다. 잡트래커는 태스크트래커에게 지시할 내용을 HeartbeatResponse에 설정할 수 있습니다. 그래서 스케줄러는 HeartbeatResponse에 태스크 실행을 요청합니다. 참고로 HeartbeatResponse에는 태스크 실행, 태스크 종료, 잡 종료, 태스크트래커 초기화 재실행, 태스크 완료 등의 작업을 설정할 수 있습니다.

4. 태스크 실행

태스크트래커는 할당받은 태스크를 새로운 JVM에서 실행하게 되며, 맵리듀스는 이를 Child JVM이라고 표현합니다. 이때 새로운 JVM에서 발생하는 버그는 태스크트래커에게 영향을 끼치지 않아서 안정적으로 태스크트래커를 운영할 수 있습니다. 또한 사용자가 원할 경우 매번 JVM을 새로 생성하는 것이 아니라 재사용하게 할 수도 있습니다.

1. TaskLauncher는 HeartbeatResponse에서 태스크 정보를 꺼내서 태스크의 상태와 진행 과정을 모니터링할 수 있는 TaskInProgress를 생성합니다.

2. 태스크트래커는 HDFS에 저장된 잡 공통 파일들을 로컬 디렉터리로 복사합니다. 그리고 TaskInProgress는 태스크 실행 결과를 저장할 로컬 디렉터리를 생성한 후 잡 JAR 파일을 이 디렉터리에 풀어 놓습니다.

3. TaskInProgress는 TaskRunner에게 태스크 실행을 요청합니다.

4. TaskRunner는 JvmManager에게 차일드 JVM에서 태스크를 실행해줄 것을 요청합니다.

5. JvmManager는 실행할 클래스명과 옵션을 설정한 후, 커맨드 라인에서 차일드 JVM을 실행합니다. 이때 차일드 JVM은 TaskUmbilicalProtocol 인터페이스로 부모 클래스와 통신하게 됩니다. 차일드 JVM은 태스크가 완료될 때까지 태스크의 진행 과정을 주기적으로 JvmManager에게 알려줍니다. 태스크트래커는 이 정보를 공유받아 태스크의 진행 과정을 모니터링 가능합니다.

6. 사용자가 정의한 매퍼 클래스 혹은 리듀서 클래스가 실행됩니다.

5. 잡 완료

1. 태스크트래커가 잡트래커게 전송하는 하트비트에는 완료된 태스크의 정보가 포함됩니다.

2. 잡트래커는 해당 잡이 실행한 전체 태스크의 완료 정보를 받게 될 경우 JobInProgress는 잡의 상태를 SUCCEEDED로 변경합니다. 만약 장애 때문에 잡이 실패했다면 잡의 상태를 FAILED로 변경합니다.

3. 잡을 실행한 클라이언트와 JobClient는 잡이 완료될 때까지 대기하고 있으며, JobClient는 잡트래커의 getJobStatus 메서드를 호출해 잡의 상태를 연속해서 확인합니다. JobClient는 잡의 상태가 SUCCEED면 true, FAILED면 false를 클라이언트에게 전달합니다.

4. 클라이언트는 최종 결과를 출력하고, 잡 실행을 완료합니다.