[하둡 완벽 가이드] - PART2 07장 맵리듀스 작동 방법

이 장에서는 하둡에서 맵리듀스의 작동 방법에 대해 자세히 살펴본다.여기서 배운 지식은 맵리듀스 고급 프로그램 작성에 좋은 밑거름이 될것이다.

7-1 맵리듀스 잡 실행 상세분석

Job 객체의 submit () 메서드 호출로 맵리듀스 잡을 실행할 수 있다.(또한 waitForCompletion() 메서드를 호출할 수 있는데, 이 메서드는 아직 잡이 제출되지 않았다면 잡을 제출하고 종료 할 때까지 기다린다.)

이 메서드가 실행되는 이면에는 엄청난 처리 과정이 숨겨저 있다. 

 

아래 그림은 잡의 전체 과정을 보여준다. 상위 수준에서 보면 다섯 개의 독립적인 단계로 구분되어 있다.

• 클라이언트: 맵리듀스 잡을 제출한다.
• YARN 리소스 매니저: 클러스터 상에 계산 리소스의 할당을 제어한다.
• YARN 노드 매니저: 클러스터의 각 머신에서 계산 컨테이너를 시작하고 모니터링 한다.
• 맵리듀스 애플리케이션 마스터:  맵리듀스 잡을 수행하는 각 태스크를 제어한다. 애플리케이션 마스터와 맵리듀스 태스크는 컨테이너 내에서 실행되며, 리소스 매니저는 잡을 할당하고 노드 매니저는 태스크를 관리하는 역할을 맡는다.
• 분산 파일시스템: 다른 단계 간에 잡 리소스 파일들을 공유하는 데 사용된다(보통 HDFS를 사용, 3장 참조).

7-1-1 잡 제출

Job의 submit() 메서드는 내부의 JobSubmitter 인스턴스를 생성하고 submitJobInternal() 메서드를 호출한다. 일단 잡을 제출하면 waitForCompletion() 메서드가 1초에 한번 씩 잡의 진행 상황을 조사하여 변경 내역이 있으면 콘솔로 보여준다. 잡이 성공적으로 완료되면 잡 카운터를 보여준다. 실패하면 잡 실패의 원인이 된 에러를 콘솔에 보여준다.

 

JobSubmitter의 잡 제출 과정은 다음과 같다.

• 리소스 매니저에 맵리듀스 잡 ID로 사용될 새로운 애플리케이션 ID를 요청한다 (2단계) .
• 잡의 출력 명세를 확인한다. 예를들어 출력 디렉터리가 지정되지 않았거나 이미 존재한다면 해당 잡은 제출되지 않고 맵리듀스 프로그램에 에러를 전달한다.
• 잡 실행에 필요한 잡 JAR 파일, 환경 설정 파일, 계산된 입력 스플릿 등의 잡 리소스를 공유 파일시스템에 있는 해당 잡 ID 이름의 디렉터리에 복사한다.(3단계) 잡 JAR 파일의 복제 인수는 mapreduce.client.submit.file.replication 속성에 정의하며 기본값은 10으로 매우 높게 설정되어 있다. 클러스터 상에 잡 JAR의 복제본이 많으면 노드 매니저가 잡의 태스크를 실행할 때 접근성이 높아지는 장점이 있다.
• 리소스 매니저의 submitApplication()을 호출하여 잡을 제출한다. (4단계)

7-1-2 잡 초기화

리소스 매니저가 submitApplication() 메서드의 호출을 받으면 YARN 스케줄러에 요청을 전달한다. 스케줄러는 컨테이너를 하나 할당하고, 리소스 매니저는 노드매니저의 운영 규칙에 따라 애플리케이션 마스터 프로세스를 시작한다. (5a와 5b 단계)

 

맵리듀스 잡의 애플리케이션 마스터는 자바 애플리케이션이며 메인 클래스는 MRAppMaster다. 애플리케이션 마스터는 잡을 초기화할 때 잡의 진행 상태를 추적하기 위한 다수의 북키핑(장부) 객체를 생성하고, 이후 각 태스크로부터 진행 및 종료 보고서를 받는다(6단계), 다음으로 클라이언트가 계산한 입력 스플릿 정보를 공유 파일시스템에서 읽어온다(7단계), 그러고 나서 입력 스플릿별로 맵 태스크 객체를 생성하고, Job의 setNumReducetask()메서드로 지정한 mapreduce.job.reduces 속성의 값(리듀서 수)만큼 맵 태스크 객체를 생성한다.이 시점에 각 태스크는 ID를 부여 받는다.

 

애플리케이션 마스터는 맵리듀스 잡을 구성하는 태스크를 실행할 방법을 결정해야 한다. 잡의 크기가 작다면 애플리케이션 마스터는 태스크를 자신의 JVM에서 실행할 수도 있다. 이러한 상황에 적합한 경우는 병렬 처리를 위해 새로운 컨테이너에 태스크를 할당하고 실행하는 오버헤드가 단일 노드에서 순차적으로 실행하는 방식에 비해 유리하다고 판단될 때다. 이러한 잡을 우버 되었다'uberized'고 말하며, 우버 태스크로 실행된다고 하기도 한다.

 

어느정도 돼야 잡이 작다고 할 수 있을까? 기본적으로 작은 잡이란 10개 미만의 매퍼와 하나의 리듀서, HDFS 블록 하나보다 작은 크기의 입력을 말한다(각각 mapreduce.job.ubertask.maxmaps, mapreduce.job.ubertask.maxreduces, mapreduce.job.ubertask.maxbytes 속성으로 설정하여 변경할 수 있다).

 

우버 태스크는 반드시 mapreduce.job.ubertask.enable 속성을 true로 변경하여 명시적으로 활성화해야 한다(개별 잡 또는 클러스터에 적용 가능).

 

마지막으로, 태스크를 실행하기 전에 애플리케이션 마스터는 OutputCommitter의 setupJob()메서드를 호출한다. 기본 클래스인 FileOutputCommitter는 잡의 최종 출력 디렉터리와 태스크 출력을 위한 임시 작업 공간을 생성한다.

7-1-3 태스크 할당

잡을 우버 태스크로 실행하기 적합하지 않다면 애플리케이션 마스터는 리소스 매니저에 잡의 모든 맵과 리듀스 태스크를 위한 컨테이너를 요청한다(8단계). 맵 태스크 요청이 먼저며 리듀스태스크 요청보다 우선순위가 높다. 리듀스의 정렬 단계가 시작되기 전에 모든 맵 태스크가 완료되어야 하기 때문이다. 전체 맵 태스크의 5%가 완료되기 전까지 리듀스 태스크의 요청은 처리되지 않는다.

 

리듀스 태스크는 클러스터의 어느 곳에서도 실행될 수 있지만, 맵 태스크 요청은 스케줄러가 최대한 준수하는 데이터 지역성'data locality' 제약이 있다. 최적의 상황은 태스크가 데이터 로컬'data local'일때다. 즉, 입력 스플릿이 저장된 노드에서 맵 태스크가 실행되는 것을 의미한다. 대안으로는 랙 로컬'rack local'이 있다.이는 동일한 랙에 속한 노드에서 맵 태스크가 실행되지만 입력 스플릿이 있는 노드는 아닐 때다. 어떤 태스크는 데이터 로컬이나 랙 로컬 어느 쪽도 아니어서 다른 랙에서 데이터를 가져오기도 한다. 개별 잡에 대한 잡 카운터를 살펴보면 지역성 수준별로 몇 개의 태스크가 실행되었는지 확인할 수 있다.

 

또한 요청할 때 태스크를 위한 메모리 요구사항과 CPU 수를 명시한다. 기본적으로 1.024MB의 메모리와 가상 코어 1개를 각 맵과 리듀스 태스크에 할당한다.이 값은mapreduce.map.memory.mb,mapreduce.reduce.memory.mb,mapreduce.map.cpu.vcores, mapreduce.reduce.cpu.vcores 속성을 통해 잡 단위로 설정할수 있다.

7-1-4 태스크 실행

리소스 매니저의 스케줄러가 특정 노드 상의 컨테이너를 위한 리소스를 태스크에 할당하면 애플리케이션 마스터는 노드 매니저와 통신하며 컨테이너를 시작한다(9a와 9b 단계). 각 태스크는 YarnChild 메인 클래스를 가진 자바 애플리케이션으로 실행된다. 태스크를 실행하기 전에 잡 환경 설정, JAR 파일, 분산 캐시와 관련된 파일(10단계, 9.4.2절 '분산 캐시' 참조) 등 필요한 리소스를 로컬로 가져와야 한다. 최종적으로 맵과 리듀스 태스크를 실제 실행한다(11단계).

 

YarnChild는 전용JVM에서 실행된다. 따라서 사용자 정의 맵과 리듀스 함수(심지어 YarnChild 내에서)에서 버그가 발생하여 강제 종료 되거나 멈추어도 노드 매니저는 영향을 받지 않는다.

 

각 태스크는 태스크 자체와 동일한 JVM에서 설정과 커밋 동작을 수행하며, 잡의 OutputCommitter가 이를 결정한다. 파일 기반의 잡에서 커밋 동작은 임시 위치에서 최종 위치로 태스크 출력을 옮긴다. 커밋 프로토콜은 투기적 실행이 활성화되었을 때 중복 태스크 중 단 하나만 커밋하고 나머지는 버린다.

스트리밍

스트리밍은 사용자가 제공한 실행 파일을 시작하고 이와 통신하기 위한 목적을 가진 특별한 맵과 리듀스 태스크를 실행한다.

 

스트리밍 태스크는 표준 입출력 스트림을 통해 프로세스와 통신한다. 태스크가 실행하는 동안 자바 프로세스는 키-값 쌍을 외부 프로세스에 전달하고,이를 사용자 정의 맵과 리듀스 함수로 처리한 뒤, 최종적으로 출력 키-값 쌍을 자바 프로세스에 돌려준다. 따라서 노드 매니저 관점에서는 자식 프로세스가 마치 맵과 리듀스 코드를 스스로 실행한 것처럼 보일 것이다.

 

7-1-5 진행 상황과 상태 갱신

맵리듀스 잡은 수행 시간이 오래 걸리는 배치 잡으로, 수십 초에서 길게는 수 시간이 소요될 수 있다.이렇게 상당한 시간이 걸리므로 사용자가 잡의 진행 상황에 대한 피드백을 받는 것은 매우 중요하다.잡과 개별 태스크는 실행 중'running', 성공적으로 완료됨 'successfully completed', 실패 'failed'와 같은 잡 또는 태스크의 상태, 맵과 리듀스의 진행 상황, 잡의 카운터 값, 상태 메세지 또는 명세(사용자 코드에서 설정된)등의 상태 정보를 가진다. 이러한 상태 정보는 잡의 진행 과정에서 수시로 변경되는데 이를 클라이언트에 전달할 방법은 무엇일까? 태스크가 수행되는 동안 태스크는 자신의 진행 상황(ex: 태스크 완료 비율)을 추적한다. 맵 태스크의 경우 이는 처리한 입력 데이터의 비율이다. 그리고 리듀스 태스크의 경우 조금은 복잡하지만 시스템은 리듀스가 처리한 입력 데이터의 비율을 추정할 수 있다. 이를 위해 전체 진행 과정을 총 세 부분으로 나누는데, 이는 셔플의 세 단계와 관련 있다. 예를 들어 리듀서에서 입력의 절반을 처리했으면 태스크의 진행 상황은 5/6가 된다. 이는 복사와 정렬 단계가 각각 1/3씩 완료되었고, 리듀스 단계의 절반(1/6)이 진행되었기 때문이다.

 

태스크는 수행 중에 발생하는 다양한 이벤트를 세는 여러 카운트를 가지는데, 여기에는 맵의 출력 레코드 수를 세는 것과 같은 프레임워크에 내장된 카운터나 사용자 정의 카운터가 있다.

 

맵과 리듀스 태스크가 실행되면서 자식 프로세스는 부모인 애플리케이션 마스터와 밀접한 인터페이스를 통해 통신한다.태스크는 진행 상황과 상태 정보(카운터 포함)를 집계하여 매 3초마다 이 인터페이스를 통해 애플리케이션 마스터에 보고한다.

 

리소스 매니저 웹 UI는 실행 중인 모든 애플리케이션 각각의 애플리케이션 마스터 웹 UI 링크를 보여준다. 애플리케이션 마스터 웹 UI에서 맵리듀스 잡의 진행 상황을 포함한 상세한 정보를 볼 수 있다. 잡이 진행되는 동안 클라이언트는 매초마다 애플리케이션 마스터를 폴링하여 가장 최근의 상태를 받는다. 또한 클라이언트는 Job의 getStatus() 메서드를 이용해서 JobStatus 인스턴스를 얻을 수 있으며, 이는 잡의 모든 상태 정보를 포함하고 있다.

 

7-1-6 잡 완료

애플리케이션 마스터가 마지막 태스크가 완료되었다는 통지를 받으면 잡의 상태를 '성공successful'으로 변경한다. 이제 잡이 상태 정보를 폴링하면 해당 잡이 성공적으로 완료되었음을 알게되고, 사용자에게 통지할 메세지를 출력한 뒤 waitForCompletion() 메서드가 반환된다. 이 시점에 잡 통계와 카운터가 콘솔에 출력된다.

 

또한 HTTP 잡 통지를 보내도록 설정되어 있다면 애플리케이션 마스터는 이를 수행한다. 클라이언트가 이러한 콜백을 받으려면 mapreduce.job.endnotification.url 속성을 설정하면 된다.

 

마지막으로, 잡이 완료되면 애플리케이션 마스터와 태스크 컨테이너는 작업 상태를 정리하고 OutputCommitter의 commitJob() 메서드를 호출한다. 잡 정보는 잡 히스토리 서버에 기록되므로 사용자가 원하는 시점에 조사할 수 있다.

7-2 실패

실제 환경에서는 사용자 코드의 버그 때문에 프로세스가 강제로 죽거나 서버에 장애가 발생하는 일이 빈번하다. 하둡의 가장 큰 장점은 이러한 실패를 잘 다루어 성공적으로 잡이 완료되도록 도와준다는 것이다. 이제 태스크, 애플리케이션 마스터, 노드 매니저, 리소스 매니저의 실패에 대해 고려해보자.

7-2-1 태스크 실패

태스크 실패 사례를 고려해보자. 가장 흔한 실패의 유형은 맵 또는 리듀스 태스크 내 사용자 코드에서 런타임 예외를 던질때다.

예외가 발생하면 태스크 JVM은 종료하기 전에 부모인 애플리케이션 마스터에 에러를 보고한다. 이 에러는 최종적으로 사용자 로그에 기록된다. 애플리케이션 마스터는 이 태스크 시도를 실패로 표시하고 해당 리소스를 다른 태스크에서 사용 가능하도록 컨테이너를 풀어준다.

 

스트리밍 태스크에서는 스트리밍 프로세스가 0이 아닌 코드를 반환하면 실패로 표시한다. 이러한 동작은

stream.non.zero.exit.is.failure 속성 (기본값은 true)에 따라 관리된다.

 

다른 실패 유형은 태스크 JVM이 갑작스럽게 종료하는 것인데, 아마도 맵리듀스 사용자 코드에 의해 드러난 특정 상황으로 인해 JVM이 종료되는 JVM 버그일 것이다. 이때 노드 매니저는 프로세스가 종료되었음을 알게 되고, 이를 애플리케이션 마스터에 알려주어 해당 시도가 실패했다고 표시하게 된다.

 

행이 걸린 태스크는 이와 다르게 처리된다. 애플리케이션 마스터는 잠시 동안 진행 상황을 갱신받지 못함을 알게 되면 해당 태스크를 실패로 표시한다. 태스크 JVM 프로세스는 이 기간 후에 자동으로 강제 종료된다. 태스크를 실패로 간주하는 타임아웃 기간은 보통 10분 이고 잡 단위(또는 클러스터 단위)로 mapreduce.task.timeout 속성에 밀리초 단위의 값을 설정할 수 있다.

 

타임아웃을 0으로 설정하면 타임아웃을 비활성화하며 따라서 실행 시간이 긴 태스크는 절대로 실패로 표시되지 않는다. 이러한 상황에서 행이 걸려 멈춘 태스크는 자신의 컨테이너를 결코 해제하지 않을 것이며 시간이 지남에 따라 클러스터를 느리게 만드는 결과를 초래한다.그러므로 이러한 접근 방법은 지양해야 하며 태스크가 주기적으로 진행 상황을 확실히 보고하도록 하는 것이 좋다.

 

애플리케이션 마스터는 태스크 시도 실패를 알게 되면 해당 태스크 실행을 다시 스케줄링 한다. 물론 애플리케이션 마스터는 이전에 실패했던 노드 매니저에 해당 태스크를 다시 스케줄링하는 것을 피하려고 노력할 것이다. 또한 태스크가 네 번 실패하면 재시도하지 않는다. 이는 설정 가능한 값으로, 맵 태스크는 mapreduce.map.maxattempts 속성으로, 리듀스 태스크는 mapreduce.reduce.maxattempts 속성으로 최대 시도 횟수를 조절할 수 있다. 기본적으로 태스크가 네 번(또는 설정한 최대 시도 횟수만큼) 실패하면 전체 잡이 실패한 것이다.

 

어떤 애플리케이션은 몇몇 태스크가 실패하더라도 잡이 중단하길 원하지 않을 수 있는데, 이는 이러한 실패에도 불구하고 잡의 실행 결과를 사용할 수 있을지 모르기 때문이다.이러한 사례에서는 잡의 실패를 유발하지 않으면서 허용되는 태스크 실패의 최대 비율을 잡에 설정할 수 있다. 맵과 리듀스 태스크는 각각 mapreduce.map.failures.maxprecent와 mapreduce.reduce.failures.maxprecent 속성으로 조절할 수 있다. 또한 태스크 시도는 강제로 종료 될 수 있는데 이는 실패와 다르다. 태스크 시도는 투기적 중첩때문에, 또는 실행 기반인 노드 매니저가 실패해서 애플리케이션 마스터가 해당 노드 매니저에서 실행되는 모든 태스크 시도를 실패로 표시할 때 강제 종료된다. 강제 종료된 태스크 시도는 태스크 자체의 잘못이 아니므로 태스크 전체 시도 횟수(mapreduce.map.maxattempts와 mapreduce.reduce.maxattempts에서 지정한)에 포함되지 않는다.

 

사용자는 웹 UI나 명령행(옵션을 보려면 mapred job을 입력하라)을 통해 태스크 시도를 강제 종료하거나 실패하게 만들 수 있다. 마찬가지로 동일한 방식으로 잡을 강제 종료할 수 있다.

7-2-2 애플리케이션 마스터 실패

맵리듀스 태스크가(하드웨어나 네트워크 실패를 직면하면) 성공하기까지 몇 번의 시도를 거치는 것처럼 YARN 내에 애플리케이션 또한 실패할 때 몇 번의 재시도를 하게 된다. 맵리듀스 애플리케이션 마스터가 시도하는 최대 횟수는 mapreduce.am.max-attempts 속성으로 조절한다. 기본값은 2며 따라서 맵리듀스 애플리케이션 마스터가 두 번 실패하면 더 이상 시도하지 않고 잡이 실패로 끝난다.

 

YARN은 클러스터에서 실행 중인 모든 YARN 애플리케이션 마스터에 대해 일괄적으로 최대 시도 횟수의 제한을 줄 수 있으며 개별 애플리케이션은 이를 넘길 수 없다.이 제한은 yarn.resourcemanager.am.max-attempts로 설정할 수 있으며 기본값은 2다. 따라서 맵리듀스 애플리케이션 마스터의 시도 횟수를 늘리고 싶다면 클러스터의 YARN 설정 또한 증가시켜야 한다. 

 

복구 작업 방식은 다음과 같다. 애플리케이션 마스터는 주기적으로 리소스 매니저에 하트비트(heartbeat)를 보내고, 애플리케이션 마스터의 실패 이벤트 발생 시 리소스 매니저는 이 실패를 감지하고 새로운 컨테이너(노드 매니저가 운영하는)에서 실행할 새로운 마스터 인스턴스를 시작한다. 맵리듀스 애플리케이션 마스터의 사례에서는 잡 히스토리를 사용해서(실패한) 애플리케이션에서 이미 실행된 모든 태스크의 상태를 복구하므로 이들을 재실행할 필요가 없다. 기본적으로 복구는 활성화되어 있지만 yarn.app.mapreduce.ap.job.recovery.enable 속성을 false로 설정하면 비활성화 된다.

 

맵리듀스 클라이언트는 진행 상황 보고를 위해 애플리케이션 마스터를 폴링하는데 만약 이 애플리케이션 마스터가 실패한다면 클라이언트는 새로운 애플리케이션 마스터 인스턴스의 위치를 알아내야 한다.처음에 잡을 초기화하는 동안 클라이언트는 리소스 매니저에 애플리케이션 마스터의 주소를 요청하고 이를 캐시하여 애플리케이션 마스터를 폴링할 필요가 있을 때마다 리소스 매니저에 요청하는 부담을 없앤다.그러나 만약 애플리케이션 마스터가 실패한다면 클라이언트는 상태 갱신을 전달할 때 타임아웃이 발생할 것이며, 이때 클라이언트는 리소스 매니저에 새로운 애플리케이션 마스터의 주소를 요청한다. 이 과정은 사용자 모르게 수행된다.

7-2-3 노드 매니저 실패

노드 매니저가 크래시에 의해 실패하거나 굉장히 느리게 수행 중이라면 리소스 매니저에 하트비트 전송을 중단할 것이다

(혹은 굉장히 드물게 전송한다) 리소스 매니저는 하트비트 전송을 중단한 노드 매니저가 10분(yarn.resourcemanager.nm.liveness-monitor.expiry-interval-ms 속성으로 밀리초 단위로 설정 가능) 동안 한번도 전송하지 않음을 인지하면 이를 컨테이너를 스케줄링 하는 노드 풀에서 제거한다.

 

실패한 노드 매니저에서 수행 중인 애플리케이션 마스터나 태스크는 앞의 두 절에서 기술한 메커니즘을 통해 복구될 것이다. 또한 애플리케이션 마스터는 해당 노드 매니저에서 성공적으로 실행된 맵 태스크가 완료되지 않은 잡에 속해있다면 이를 재실행한다.

이는 노드 매니저의 로컬 파일 시스템에 존재하는 중간 결과를 리듀스 태스크에서 접근할 수 없기 때문이다.

 

애플리케이션 실패 횟수가 높으면 노드 매니저 자체가 실패하지 않았더라도 노드 매니저는 블랙리스트에 등록된다. 블랙리스트 등록은 애플리케이션 마스터가 하며, 하나의 노드 매니저에서 네 개 이상의 맵리듀스 태스크가 실패하면 다른 노드에 태스크를 다시 스케줄링 할 것이다. 사용자는 mapreduce.job.maxtaskfailures.per.tracker 잡 속성으로 이 한계 값을 설정할 수 있다.

7-2-4 리소스 매니저 실패

리소스 매니저 실패는 굉장히 심각한 상황이다. 리소스 매니저 없이는 잡이나 태스크 컨테이너가 실행될 수 없기 때문이다. 기본 환경 설정에서 서버 실패 이벤트 발생시(거의 발생하지 않치만) 모든 실행 중인 잡은 실패하며 복구 불가하기 때문에 리소스 매니저는 단일 고장점이다.

 

고가용성(HA)을 달성하기 위해서는 두개의 리소스 매니저를 활성 대기 설정으로 실행해야 한다. 만약 활성 리소스 매니저가 실패해도 대기 리소스 매니저가 그 역할을 대신하기 때문에 클라이언트는 심각한 방해를 받지 않는다.

 

실행중인 애플리케이션에 대한 모든 정보는 고가용 상태 저장소(주키퍼 혹은 HDFS)에 보관되기 때문에 대기 리소스 매니저는 실패한 활성 리소스 매니저의 핵심 상태를 복구할 수 있다.그리고 노드 매니저 정보는 상태 저장소에 보관되지 않는데, 이는 노드 매니저가 첫번째 하트비트를 전송할 때 새로운 리소스 매니저가 상대적으로 빠르게 재구축 할 수 있기 때문이다.

 

새로운 리소스 매니저가 시작되면 상태 저장소로 애플리케이션 정보를 읽고 클러스터에서 실행 중인 모든 애플리케이션의 애플리케이션 마스터를 재시작한다. 애플리케이션 코드 에러로 인해 실패한 것이 아니라 시스템에 의해 강제로 종료되었으므로 이를 실패한 애플리케이션 시도로 세지 않는다. 실제 환경에서 애플리케이션 마스터 재시작 시 맵리듀스 애플리케이션은 완료된 태스크의 작업을 복구할 수 있으므로 큰 문제가 되지 않는다.

 

대기 리소스 매니저에서 활성 리소스 매니저로의 전환은 장애극복 관리자가 담당한다.기본 장애극복 관리자는 주키퍼 대표자 선출을 사용해서 어떠한 시점에라도 단일 활성 리소스 매니저가 존재하도록 보장한다. HDFS 고가용성과 달리 장애극복 관리자는 반드시 독립 프로세스일 필요는 없으며 편리한 환경설정을 위해 기본적으로 리소스 매니저에 포함된다. 물론 수동으로 장애극복 설정이 가능하지만 권장하지 않는다.

 

이제 통신 가능한 리소스 매니저가 두 개 존재하므로 클라이언트와 노드 매니저는 리소스 매니저의 장애극복을 처리할 수 있도록 설정되어야 한다. 즉, 활성 리소스 매니저를 찾을 때까지 라운드 로빈 방식으로 각 리소스 매니저에 연결을 시도해야 한다. 만약 활성 리소스 매니저가 실패한다면 대기 리소스 매니저가 활성화 될 때까지 재시도 할것이다.

7-3 셔플과 정렬

맵리듀스는 모든 리듀서의 입력이 키를 기준으로 정렬되는 것을 확실히 보장한다. 시스템이 이러한 정렬을 수행하고 맵의 출력을 리듀서의 입력으로 전송하는 과정을 셔플이라고 한다. 이 절에서는 셔플의 작동 방식을 살펴볼 것이며, 이러한 기본적인 이해를 통해 맵리듀스 프로그램을 최적화하는데 큰 도움을 얻을 것이다.셔플은 정제와 개선이 끊임없이 일어나는 코드 기반 영역이므로 이어지는 설명에서 자세한 부분은 필요에 따라 생략할 수 있다. 여러모로 셔플은 맵리듀스의 핵심이며 '마법'이 일어나는 곳이다.

7-3-1 맵부분

맵 함수가 결과를 생산할 때 이를 단순히 디스크에 쓰지 않는다. 좀더 복잡한 처리과정이 있으며 효율적인 처리를 위해 메모리에 일정 크기만큼 쓴 다음 사전 정렬을 수행한다.

 

각 맵 태스크는 환형 구조의 메모리 버펴를 가지고 있으며 이곳에 결과를 기록한다.이 버퍼는 기본적으로 100MB 크기다(mapreduce.task.io.sort.mb 속성으로 변경 가능). 버퍼의 내용이 특정 한계치 (mapreduce.map.sort.spill.percent 속성으로 변경 가능하며 기본값은 0.80 또는 80%)에 도달하면 백그라운드 스레드가 디스크에 스필하기 시작한다. 스필이 일어나는 동안에도 맵 결과는 계속해서 버퍼에 쓰이는데 이때 버퍼가 가득 차게 되면 맵은 스필이 종료 될 때까지 블록된다.스필은 잡의 특정 서브디렉터리 내에 mapreduce.cluster.local.dir 속성으로 지정한 디렉터리에 라운드 로빈 방식으로 쓰인다.

 

디스크로 쓰기 전에 스레드는 먼저 데이터를 최종적으로 전송할 리듀서 수에 맞게 파티션으로 나눈다. 각 파티션 내의 백그라운드 스레드는 키를 기준으로 인메모리 정렬을 수행하고 만약 컴바이너 함수가 존재하면 정렬의 출력에 대해 수행한다. 컴바이너 함수 수행은 맵 출력을 더욱 축소하여 로컬 디스크에 쓰거나 리듀서에 전송할 데이터 양을 줄인다.

 

메모리 버퍼가 스필 한계치에 도달하면 새로운 스필 파일이 생성되므로 맵 태스크가 최종결과 레코드를 쓰고 나면 여러개의 스필파일이 존재할 수 있다.태스크가 종료되기 전에 여러 스필 파일은 단일 출력 파일로 병합되고 정렬된다. mapreduce.task.io.sort.factor 환경 설정 속성은 한번에 병합할 최대 스트림 수를 조절하며 기본값은 10이다.

 

최소 세 개의 스필파일(mapreduce.map.combine.minspills 속성으로 설정)이 존재한다면 출력 파일을 쓰기 전에 컴바이너가 다시 실행된다. 컴바이너는 최종 결과에 영향을 미치지 않으면서 입력에 대해 반복적으로 수행될 수 있음을 상기하자.하나 혹은 두 개의 스필만 존재한다면 맵 출력 크기를 줄이는 양에 비해 컴바이너를 호출하는 오버헤드가 크기 때문에 이러한 맵 출력에 대해 다시 실행하는 것은 가치가 없다.

 

맵 출력을 디스크에 쓰려는 시점에 압축하는 것은 디스크에 더욱 빨리 쓰고 디스크 공간을 절약하며 리듀서로 전송할 데이터 양을 줄일 수 있으므로 때론 훌륭한 생각이다.기본적으로 출력은 압축되지 않지만 mapreduce.map.output.compress 속성을 true로 설정하면 이를 간단히 활성화 할 수 있다. 사용할 압축 라이브러리는 mapreduce.map.output.compress.codec에 지정한다.

 

출력 파일의 파티션은 HTTP를 통해 리듀서에 전달된다. 여기서 파일 파티션을 전달하는 워커 스레드의 최대 수는 mapreduce.shuffle.max.threads 속성으로 조절 가능하다. 이 설정은 맵 태스크가 아닌 노드 매니저별로 이루어지며, 기본값인 0은 스레드의 최대 수를 서버 프로세서 수의 두 배로 설정함을 의미한다.

7-3-2 리듀스 부분

이제 리듀스 부분에서의 처리로 넘어가자. 맵 출력 파일은 맵 태스크를 수행한 서버의 로컬 디스크에 존재하는데(맵의 출력은 항상 로컬 디스크에 쓰이지만 리듀스의 출력은 그렇지 않음을 명심하라), 이제는 이것이 특정 파티션에 대해 리듀스 태스크를 시작하려는 서버에서 필요한 상황이다. 게다가 리듀스 태스크는 클러스터 내에 퍼저 있는 많은 맵 태스크로부터 특정 파티션에 해당하는 맵 출력을 필요로 한다. 맵 태스크는 각기 다른 시간에 끝날 수 있으므로 리듀스 태스크는 각 맵 태스크의 출력이 끝나는 즉시 복사하기 시작한다. 이것이 리듀스 태스크의 잘 알려진 복사단계다. 리듀스 태스크는 소수의 복사기 스레드를 가지고 있는데, 이들은 맵 출력 인출을 병렬로 수행한다.

 

기본적으로 다섯 개의 스레드가 설정되어 있지만, 이 수는 mapreduce.reduce.shuffle.parallelcopies 속성으로 변경할 수 있다.

맵 출력은 그 크기가 충분히 작다면 리듀스 태스크 JVM 메모리에 복사된다(메모리 버퍼의 크기는 이러한 목적으로 사용할 힙의 비율을 지정하는 mapreduce.reduce.shuffle.input.buffer.percent 속성으로 조절한다). 인메모리 버퍼가 한계치(mapreduce.reduce.shuffle.merge.percent로 조절)에 도달하거나 맵 출력수가 한계치(mapreduce.reduce.merge.inmem.threshold로 조절)에 도달하면 병합되어 디스크에 스필된다. 컴바이너가 지정되었다면 병합 도중에 실행되어 디스크에 쓰여질 데이터 양을 줄일 수 있다.

 

복사된 파일이 디스크에 축적되면 백그라운드 스레드가 이를 더 크고 정렬된 형태의 파일로 병합한다.이는 추후에 병합할 시간을 절약해준다.(맵 태스크에 의해) 압축된 맵 출력이라면 병합을 수행하기 위해 메모리 내에서 압축을 풀어야 한다.

 

모든 맵 출력이 복사되는 시점에 리듀스 태스크는 정렬 단계(정렬은 맵 단계에서 수행되었기 때문에 병합 단계라고 하는 것이 더 적절하다)로 이동하여 맵 출력을 병합하고 정렬 순서를 유지한다. 이 작업은 라운드 단위로 이루어진다. 예를 들어 50개의 맵 출력이 존재하고 병합 계수 가 10(이는 기본값이며 맵의 병합과 마찬가지로 mapreduce.task.io.sort.factort 속성으로 조절)이라면 다섯 개의 라운드로 구성될 것이다. 각 라운드는 10개의 파일을 하나로 병합하여 결과적으로 다섯 개의 중간 파일이 생성된다.

 

이 다섯 개의 파일을 하나의 정렬된 파일로 병합하는 최종 라운드를 가지는 대신 마지막 단계(리듀스 단계) 의 리듀스 함수에 곧바로 전송하여 디스크 IO를 줄인다. 최종 병합은 메모리와 디스크 세그먼트의 혼합으로 이루어진다.

리듀스 단계에서 리듀스 함수는 정렬된 출력 내의 각 키에 대해 호출된다. 이 단계의 출력은 보통 HDFS와 같은 출력 파일시스템에 곧바로 써진다. HDFS에서 노드 매니저는 데이터노드를 실행하므로 블록의 첫번째 복사본은 로컬디스크에 쓰일 것이다.

 

7-3-3 설정 조정

이제 맵리듀스 성능 향상을 위한 셔플 튜닝 방법을 이해하기 더 나은 위치에 와 있다. 이와관련된 설정은 기본값과 함께 요약되어 있다.

- 맵 측면에서 튜닝속성

속성명	타입	기본값	설명
mapreduce.task.io.sort.mb	int	100	맵 출력을 정렬하는 동안 사용할 메모리 버퍼의 크기로,메가바이트 단위
mapreduce.map.sort.spill.percent	float	0.80	디스크로의 스필을 시작하기 위한 맵 출력 메모리 버퍼와 레코드 경계 인덱스의 한계 사용 비율
mapreduce.task.io.sort.factor	int	10	파일 정렬시 한번에 병합할 스트림의 최대수. 이 속성은 리듀스에서도 사용된다 100으로 증가시키는 것이 꽤 일반적이다.
mapreduce.map.combine.minspills	int	3	컴바이너를 실행하기 위해 필요한 스필 파일의 최소수
mapreduce.map.output.compress	boolean	false	맵 출력 압축 여부
mapreduce.map.output.compress.codec	Class	org.apache.hadoop.io.compress.DefaultCodec	맵 출력에 사용할 압축 코드
mapreduce.shuffle.max.threads	int	0	맵 출력을 리듀서에 제공하기 위한 노드 매니저 별 워커 스레드 수. 클러스터 전체 설정이며 개별 잡에 설정이 불가능하다. 0은 네티 기본값인 가용 프로세서 수의 두배다.

 

- 리듀스 측면에서 튜닝속성

속성명	타입	기본값	설명
mapreduce.reduce.shuffle.parallelcopies	int	5	맵 출력을 리듀서에 복사하기 위해 사용되는 스레드 수
mapreduce.reduce.shuffle.maxfetchfailures	int	10	리듀서가 에러를 보고하기 전에 수행하는 맵 출력 인출 시도수
mapreduce.task.io.sort.factor	int	10	파일을 정렬할 때 한번에 병합하는 스트림의 최대수. 이 속성은 맵에서도 사용된다
mapreduce.reduce.shuffle.input.buffer.percent	float	0.70	셔플의 복사 단계 동안 맵 출력 버퍼에 할당되는 전체 힙 크기 비율
mapreduce.reduce.shuffle.merge.percent	float	0.66	출력의 병합과 디스크에 스필을 시작하기 위한 맵 출력 버퍼(mapred.job.shuffle.input.buffer.percent로 정의)의 한계 사용 비율
mapreduce.reduce.merge.inmem.threshold	int	1000	출력의 병합과 디스크에 스필하는 과정을 시작하기 위한 맵 출력의 한계 수. 0 이하의 값은 한계가 없다는 의미며, 스필 동작은 mapreduce.reduce.shuffle.merge.percent에 의해 결정된다.
mapreduce.reduce.input.buffer.percent	float	0.0	리듀스가 진행되는 동안 맵 출력을 메모리에 유지하는데 사용되는 전체 힙 크기의 비율. 리듀스 단계가 시작되면 메모리 내의 맵 출력 크기는 이 크기를 넘을 수 없다. 기본적으로 리듀스에 가능한 한 많은 메모리를 할당하기 위해 리듀스를 시작하기 전에 모든 맵 출력을 디스크에 병합해 놓는다. 하지만 리듀서가 요구하는 메모리가 적다면 디스크 IO를 최소화 하기 위해 이 값은 증가될 수 있다.

 

일반적인 원칙은 셔플에 가능한 한 많은 메모리를 할당하는 것이다. 그러나 여기에는 트레이드 오프가 존재하는데, 맵과 리듀스 함수가 동작하는데 충분한 메모리 확보가 필요하기 때문이다. 따라서 맵과 리듀스 함수를 작성할 때 가능한 한 적은 메모리를 사용하도록 하는 것이 최선책이다.맵과 리듀스에서 무한정으로 메모리를 사용하지 않도록 확실히 하자(예를 들어 맵에 값을 축적하는 것을 피하자)

 

맵과 리듀스 태스크를 실행하는 JVM에 할당된 메모리 크기는 mapred.child.java.opts 속성으로 설정한다. 이때 태스크 노드에 가능한 한 많은 양의 메모리를 할당하는 것이 좋다. 

 

맵 측면에서 보면 다수의 디스크 스필을 피하는 것(하나가 최선이다)이 최고의 성능을 내는 방법이다. 맵 출력 크기를 측정할 수 있다면mapreduce.task.io.sort.* 속성을 적절히 설정하여 스필 횟수를 최소화 할 수 있다. 가능하면 mapreduce.task.io.sort.mb를 늘릴 것을 권장한다. 잡 실행 중에 디스크에 스필되는 전체 레코드 수를 세는 맵리듀스 카운터가 존재한다면 튜닝하기 유용할 것이다. 이 카운터는 맵과 리듀스 측의 모든 스필을 포함한다.

 

리듀스 측면에서는 중간 데이터 전체가 메모리에 존재할 때 최고의 성능을 얻을 수 있다. 기본적으로 이러한 일은 발생하지 않는데, 일반적으로 리듀스 함수에 모든 메모리를 예약해 두기 때문이다. 그러나 리듀스 함수가 조금의 메모리만 요구할 때는 mapreduce.reduce.merge.inmem.threshold는 0으로, mapreduce.reduce.input.buffer.percent는 1.0(혹은 더 낮은값)으로 설정하면 성능 향상을 꾀할 수 있다.

 

좀더 일반적으로 하둡의 기본 버퍼 크기는 4KB로 작기 때문에 클러스터 전반에 걸쳐 이를 늘릴것을 권장한다.

7-4 태스크 실행

7.1절 '맵리듀스 잡 실행 상세분석'에서 맵리듀스 시스템이 전체적인 잡 과정에서 태스크를 어떻게 실행하는지 살펴보았다. 이 절에서는 맵리듀스 사용자가 태스크 실행에 관해 취할 수 있는 좀 더 많은 제어사항을 알아볼 것이다.

7-4-1 태스크 실행 환경

하둡은 맵 또는 리듀스 태스크에 실행 환경에 관한 정보를 전달해준다. 예를 들어 맵 태스크는 처리할 파일명을 알 수 있고,맵 또는 리듀스 태스크는 태스크 시도 횟수를 알 수 있다.아래 표에 있는 속성은 이전 맵 리듀스 API가 제공하는 Mapper 또는 Reducer의 configure() 메서드 구현(환경 설정 인자로 받는)을 통해 얻을수 있는 잡의 환경설정으로 접근 가능하다. 새로운 API에서 이 속성은 Mapper 또는 Reducer의 모든 메서드에 전달된 콘텍스트 객체로부터 접근할 수 있다.

-태스크 환경 속성

속성명	타입	설명	예시
mapreduce.job.id	String	잡ID	job_200811201130_0004
mapreduce.task.id	String	태스크 ID	task_200811201130_0004_m_000003
mapreduce.task.attempt.id	String	태스크 시도 ID	attempt_200811201130_0004_m_000003_0
mapreduce.task.partition	int	잡 내 태스크의 인덱스	3
mapreduce.task.ismap	boolean	태스크가 맵 태스크인지 여부	true

스트리밍 환경변수

하둡은 잡 환경 설정 파라미터를 스트리밍 프로그램의 환경변수로 설정한다.이때 알파벳이나 숫자 이외의 문자는 밑줄(_)로 대체하여 유효한 이름으로 만든다. 파이썬 구문은 파이썬 스트리밍 스크립트에서 mapreduce.job.id 속성 값을 얻어오는 방법을 보여준다.

os.environ["mapreduce_job_id"]

또한 맵리듀스가 스트리밍 프로세스 실행시 -cmdenv 옵션을 스트리밍 시작 프로그램에 입력하여 환경변수를 설정할 수 있다.(설정하려는 변수마다 한번씩 사용). 다음은 MAGIC_PARAMETER 환경변수를 설정하는 예다.

.cmdenv MAGIC_PARAMETER=abracadabra

 

7-4-2 투기적 실행

맵리듀스 모델은 잡을 태스크로 나누고 태스크를 병렬 수행하는 형태다(이는 태스크를 순차적으로 실행할 때보다 전체 잡 실행 시간을 줄여준다). 이로 인해 잡 실행 시간은 느리게 수행되는 태스크에 굉장히 민감한데, 이는 단 하나의 느린 태스크가 전체 잡 수행을 상당히 지연시키기 때문이다. 실제로 수백,수천개의 태스크로 구성된 잡에서 몇몇 태스크가 뒤처질 가능성은 매우 높다.

 

태스크는 하드웨어의 성능 저하나 소프트웨어의 잘못된 설정 등 다양한 이유로 느려질 수 있다. 하지만 기대했던 시간보다 더 오랜 시간이 걸리더라도 태스크가 성공적으로 종료되었다면 원인을 진단하기 어려울 수 있다.하둡은 느린 태스크를 진단하거나 고치려 하지 않는 대신 태스크 수행이 예상했던 것보다 더 느린 상황을 감지하여 또 다른 동일한 예비 태스크를 실행한다.이를 태스크의 투기적 실행 이라고 한다.

 

동시에 두 개의 복제 태스크를 실행하여 서로 경쟁하도록 하는 것이 투기적 실행이 아님을 명심해야 한다.이 방법은 클러스터의 리소스 낭비가 심하다.대신 스케줄러는 잡의 동일한 타입(맵과 리듀스)별로 모든 태스크의 진행 상황을 기록하고 평균보다 심각하게 느리게 수행 중인 적은 비율의 태스크에 대해 투기적 복제 태스크를 실행한다.이들 중 하나의 태스크가 성공적으로 완료되면 실행 중인 중복 태스크는 더 이상 필요 없으므로 강제 종료되고 투기적 태스크가 먼저 종료되면 원래 태스크가 강제 종료된다.

 

투기적 실행은 일종의 최적화며 더욱 안정적으로 잡을 실행하는 기능은 아니다. 따라서 태스크를 멈추거나 느리게 하는 버그가 존재할 때 이러한 문제를 회피하기 위해 투기적 실행에 의존하는 것은 옳지 않으며 안정적으로 동작하지도 않을 것이다. 동일한 버그가 투기적 태스크에도 영향을 미칠 가능성이 있기 때문이다. 그러므로 태스크가 멈추거나 느려지지 않도록 버그를 고쳐야 한다.

 

기본적으로 투기적 실행은 활성화 되어 있다. 이 설정은 클러스터 전체 혹은 잡별로 맵 태스크와 리듀스 태스크에 각각 독립적으로 활성화나 비활성화 할 수 있다. 아래 표에서는 이와 관련된 속성을 보여준다.

 

-투기적 실행 속성

속성명	타입	기본값	설명
mapreduce.map.speculative	boolean	true	어떤 태스크가 느리게 진행될 때 맵태스크의 추가 인스턴스를 실행할지 여부
mapreduce.reduce.speculative	boolean	true	어떤 태스크가 느리게 진행될 때 리듀스 태스크의 추가 인스턴스를 실행할지 여부
yarn.app.mapreduce.am.job.speculator.class	Class	org.apache.hadoop.mapreduce.v2.app.speculate.DefaultSpeculator	투기적 실행 정책을 구현한 Speculator 클래스
yarn.app.mapreduce.am.job.task.estimator.class	Class	org.apache.hadoop.mapreduce.v2.app.speculate.LegacyTaskRuntimeEstimator	태스크 런타임 측정을 제공하고 Speculator 인스턴스가 사용하는 TaskRuntimeEstimator 구현체

 

투기적 실행의 궁극적인 목적은 잡 실행 시간을 줄이는것이지만 클러스터 효율성 측면에서 비용이 발생한다. 혼잡한 클러스터에서 개별 잡의 실행시간을 줄이기 위해 투기적 실행으로 중복 태스크가 실행되면 전반적인 단위 시간당 산출물은 줄어든다. 이러한 이유로 클러스터 관리자는 클러스터 상에서 이를 끄는것을 선호하며, 사용자가 개별 잡에 대해 필요하면 명시적으로 켜도록 하고 있다. 이것은 오래된 하둡 버전에서 투기적 실행으로 투기적 태스크를 심하게 공격적으로 스케줄링 할 때 특히 문제가 되었다.

 

리듀스 태스크에 투기적 실행을 끄는 것은 좋은 예다. 복제 리듀스 태스크는 원래 태스크와 동일한 맵 출력을 인출해야 하며, 이는 클러스터에서 네트워크 트래픽을 심각하게 증가시키기 때문이다.

 

투기적 실행을 끄는 또 다른 이유는 비멱등 태스크 때문이다. 하지만 많은 사례에서 태스크를 멱등이 되도록 작성하고 OutputCommitter를 사용해서 태스크 성공 시 출력을 최종 위치로 옮길 수 있다. 이 기술은 다음 절에서 더 자세히 설명한다.

7-4-3 출력 커미터

하둡 맵리듀스는 잡과 태스크가 깨끗하게 성공하거나 실패하도록 보장하기 위한 커밋 프로토콜을 사용한다. 이 동작은 잡에서 사용중인 OutputCommitter로 구현했으며 예전 맵리듀스 API 내에 JobConf의 setOutputCommitter() 함수를 호출하거나 환경설정에서 mapred.output.committer.class 속성으로 설정할 수 있다. 새로운 맵리듀스 API에서 OutputCommitter는 OutputFormat의 getOutputCommitter() 메서드로 확인할 수 있다. 새로운 맵리듀스 API에서 OutputCommitter는 outputFormat의 getOutputCommitter() 메서드로 확인할 수 있다. 기본값은 파일 기반의 맵리듀스에 적합한 FileOutputCommitter다. 또한 기존의 OutputCommitter를 사용자화하거나 잡 또는 태스크에 특별한 설정이나 정리가 필요하다면 새로운 구현체를 작성할 수도 있다.

 

태스크의 부차적인 파일

맵과 리듀스 태스크의 출력을 작성하는 일반적인 방법은 키-값 쌍을 수직하는 OutputCollector를 사용하는 것이다. 몇몇 애플리케이션은 단일 키-값 쌍 모델보다 더 유연한 모델을 필요로 하기 때문에 맵또는 리듀스 태스크에서 직접 HDFS와 같은 분산 파일 시스템에 출력 파일을 작성한다.

 

주의해야 할 점은 동일 태스크의 다중 인스턴스가 동일한 파일에 쓰지 않도록 보장하는 것이다. 이전 절에서 보았듯이 OutputCommitter 프로토콜은 이러한 문제를 해결해준다.(책 API를 참조)

애플리케이션이 태스크 작업 디렉터리에 부차적인 파일을 작성했을 때 성공적으로 완료된 태스크의 부차적인 파일은 출력 디렉터리에 자동으로 옮겨지지만 실패한 태스크의 파일은 삭제될 것이다.

 

태스크는 잡의 환경 설정에 mapreduce.task.output.dir 속성 값을 얻어와서 작업 디렉터리를 찾을 수 있다. 또는 자바 API를 사용하는 맵리듀스 프로그램은 FileOutputFormat의 getWorkoutputPath() 정적 메서드를 호출하여 작업 디랙터리를 나타내는 Path 객체를 얻을 수 있다. 프레임 워크가 태스크를 실행하기 전에 작업 디렉터리를 생성하므로 사용자가 별도로 생성할 필요는 없다

 

예를들어 이미지 파일 포맷을 변환하는 프로그램을 상상해보자. 이를 위한 한가지 방법은 변환할 이미지 집합을 전달받는(NLineInputFormat을 사용) 맵 -단독 잡을 이용하는 것이다. 만약 맵 태스크가 변환된 이미지를 작업 디렉터리에 쓴다면 태스크가 성공적으로 종료될때 출력 디렉터리로 이동될 것이다.