[펌_Hadoop] 거침없이 하둡 활용해 보기

 

---

 출처: http://socurites.com/124

---

 

 

 

이 문서는 "거침없이 배우는 하둡"을 읽고 쓴 독후감이다.

가장 기본적인 내용은 간단히 정리해보고, 지금까지 미처 활용하지 못했던 몇가지 팁을 작성하고자 한다.

하둡이란

하둡은 빅데이터를 처리할 수 잇는 분산 응용 프로그램을 작성하고, 실행하기 위한 오픈 소스 프레임워크다.

분산 컴퓨팅 플랫폼은 하둡 외에도 다양하게 존재하지만, 하둡은 아래와 같은 차이점을 갖는다.

• 접근성(Accessibility) : 하둡은 다수의 범용 컴퓨터로 클러스터를 쉽게 구성할 수 있으며, 또는 아마존의 EC2(Elastic Compute Cloud)와 같은 클라우드 인프라를 간단히 활용할 수 있다.
• 견고함(Robust) : 하둡은 범옹 컴퓨터로 구성된 클러스터상에서 실행되도록 설계되었기 떄문에, 하드웨어가 고장이 나더라도 쉽게 복구할 수 있도록 만들어졌다.
• 확장가능성(Scalability) : 처리 속도나 저장 용량을 늘리고자 할 때, 단순히 컴퓨터를 하나 추가함으로써 선형적으로 확장이 가능하다.
• 단순함(Simplicity) : 분산된 컴퓨터에서 병렬적으로 처리되는 프로그램을 손쉽게 개발할 수 있다.

하둡의 설계 철학

일반적으로 프로그램은 데이터를 프로그램이 있는 컴퓨터에 가져와서 처리하게 된다.

반면 하둡의 경우, 데이터가 있는 컴퓨터에 프로그램을 전송해서 그 곳에서 데이터를 처리한다.

따라서 실제 데이터가 저장되어 있는 컴퓨터에서 데이터가 처리된다.

데이터가 대용량이라는 점을 감안해 볼 때, 데이터를 옮기는 것보다 프로그램을 옮기는 것이 당연하며,

하둡은 이러한 철학을 따라 만들어졌다.

MapReduce 살펴보기

MapReduce 프로그램은 map과 reduce, 2단계로 이루어진다. map 단계에서는 입력 데이터를 필터링한 후 reduce 단계에서 처리할 수 있는 형태로 변경한다. reduce는 map의 출력 값을 입력 값으로 받아서 통합하게 된다.

하둡의 구성 요소

하둡은 아래의 데몬으로 구성된다.

• NameNode
• DataNode
• Secondary NameNode
• Job Tracker
• TaskTracker

NameNode

하둡은 마스터/슬레이브 구조를 가진다. 네임노드는 HDFS에서 마스터 역할을 하며, 슬레이브 역할을 하는 데이터 노드에게 I/O 작업을 할당한다. 네임노드는 현재 이중화 구성을 할 수 없어서, 하둡 클러스터를 운영할 때 단일 장애 지점이 된다.

* 네임노드 백업 및 이중화에 대해서는 조금 후에 살펴보도록 하겠다.

DataNode

실제 데이터는 데이터노드에 저장된다. 클라이언트가 HDFS에 파일을 읽거나 쓰기 위해 네임노드에게 요청을 날리면, 네임노드는 어느 데이터노드의 어디 블록에 파일이 있는지(또는 쓸지)를 알려준다. 그러면 클라이언트는 데이터노드와 직접 통신하여, 파일을 읽거나 쓰게된다. 다시 말해, 데이터노드와 블록 위치가 정해지면 클라이언트는 네임노드와는 전혀 통신하지 않고, 해당 데이터 노드와 직접 통신한다.

클러스터가 처음 시작될 때, 각 데이터 노드에서 자신의 블록 정보를 네임노드에게 알려준다. 그런 후 데이터 노드는 자신의 로컬 디스크에 변경사항이 발상할 때마다 네임노드에게 변경사항을 알려주게된다.

Secondary Name Node

세컨데리 네임노드는 HDFS의 블록이 변경되더라도 실시간으로 변경정보를 갱신하지 않는다. 세컨데리 네임노드는 네임노드의 블록정보를 합칠 때 사용하기 위한 노드이며, 백업 노드도 아니며 stand by 네임 노드도 아니다. 

** 참고 : hadoop의 secondary namenode는 백업용이 아닙니다.

하지만 네임노드에 장애가 발생할 경우를 대비해서 백업용 노드로 활용할 수 있다. 그리고 네임노드가 장애가 난 경우, 세컨데리 네임노드를 네임노드로 활용할 수도 있다. 이러한 작업은 모두 자동화 되지 않으며, 수작업으로 처리해야 한다.

* 네임노드 백업 및 이중화에 대해서는 조금 후에 살펴보도록 하겠다.

JobTracker

클라이언트가 요청한 작업(job)에 대한 실행을 관리한다. 일반적으로 잡트래커는 마스터 노드, 즉 네임노드가 설치된 노드에서 실행한다.

TaskTracker

작업을 실제로 처리하는 일은 태스크트래커가 맡는다. 슬레이브 노드, 즉 데이터 노드에는 하나의 태스크트래커만이 존재하며, 여러개의 JVM을 생성해서 다수의 map과 reduce를 실행하게 된다.

 

Job Chaining

복잡한 문제가 있을 떄 하나의 프로그램으로 처리할 수도 있지만, 작은 책임만을 가지도록 프로그램을 나누면 프로그램을 유지보수하기가 쉬워진다. MapReduce에서도 여러개의 작은 Job을 연결하여 복잡한 문제를 처리할 수 있다.

Job Pipeline

Job A와 Job B가 있고, A 작업이 끝난 후 A 작업의 출력을 B 작업의 입력으로 사용해야한다고 해보자. 수작업으로 A 작업이 끝난뒤, B 작업을 실행할 수도 있지만, 파이프라인을 이용하면 여러개의 Job을 서로 연결할 수 있다.

job_A | job_B

이와 같이 실행한 경우, job_A의 드라이버가 실행 된 후, job_B의 드라이버가 실행된다. 그리고 job_A의 결과 파일이 job_B의 입력파일로 설정된다.

Job Dependency

여러가지 Job이 있고, 각 Job이 모두 순차적으로 실행되지는 않을 경우 단순히 파이프라인만으로는 실행 순서를 결정할 수 없다. 예를 들어 job_A, job_B, job_C가 있는 경우, job_A와 job_B는 job_C보다 먼저 실행되어야 하지만, job_A와 job_B는 어느 순서로 실행되어도 되는(또는 동시에 실행되어도 관계 없는) 경우를 가정해보자. 이 경우 아래와 같이

job_A | job_B | job_C

또는

job_B | job_A | job_C

와 같이 실행해도 원하는 결과를 얻을 수 있다. 하지만 처리속도는 최적화되지 않는데, job_A와 job_B는 동시에 실행하는 편이 낫기 때문이다. 이러한 경우 의존성을 직접 설정하여 처리 순서를 결정할 수 있다.

예를 들어 job_C에 대해 아래와 같이 설정할 수 있다.

job_C.addDependingJob(job_A);

job_C.addDependingJob(job_B);

이는 job_A와 job_B가 종류된 후, job_C가 실행되는 것을 보장한다.

이처럼 의존성이 결정된 job은 모두 JobControl 객체에 등록해야하며, 이때 addJob() 메서드를 이용한다.

의존성과 job 등록이 완료되면, JobControl.run() 메서드를 실행하여 job을 실행할 수 있다.

JobControl에는 allFinished() 메서드와, getFailedJobs() 메서드가 있어 job들의 실행상태를 모니터링할 수 있다.

ChainMapper와 ChainReducer

데이터를 처리할 때 두가지 책임을 갖는 매퍼를 만드는 경우가 있다. 예를 들어 웹로그를 읽어서 사용자 접근 유형을

분석하는 MapReduce를 개발한다고 해보자. 이경우 아래와 같은 처리흐름을 따라 처리할 수 있다.

• 로그 정규화 및 유효하지 않은 라인 제거
  입력 파일의 로그 중에서, 정해진 로그 포맷에 맞지 않는 라인을 걸러낸다.
  
• 클라이언트 Ip별로 매핑
  클라이언트 IP를 키로, 상태가 200인 접근 로그를 값으로 매핑한다.
  
• 클라이언트 IP별 접근 유형 리듀싱
  클라이언트 IP별로 접근 로그를 분석하여 결과파일로 리듀싱한다.
  
• 리듀싱된 결과 DB에 로드
  리듀싱된 결과를 DB에 로드한다.

하나의 Job으로도 위의 로직을 모두 구현할 수 있으나, 첫번쨰 유효성 검사 작업과 마지막 DB 로드 작업은 다른 Job에서도 공통적으로 필요한 모듈이므로 별도의 모듈로 관리하면 재사용이 용이하다. 따라서 아래와 같이 Job을 구조화해볼 수도 있다.

Job_A | Job_B | Job_C

Job_A

mapper : 웹로그의 입력을 읽어 정규화한다. 키는 라인번호, 값은 로그 라인 전체를 매핑한다.

reducer : IdentityReducer를 사용한다.

Job_B

mapper : 클라이언트 Ip별로 매핑. 키는 클라이언트 IP, 값은 접근 로그

reducer : 클라이언트 IP별 접근 유형 리듀싱. 키는 클라이언트 IP, 값은 분석 결과

Job_C

mapper : 분석 결과를 DB로 로드한다. 출력은 없음

reducer : 없음

이처럼 파이프라인을 활용하면 책임에 따라 Job을 하위 모듈로 구조화할 수 있지만, 각 단계마다 중간 결과를 처리해야 하므로 I/O가 발생하게 된다.

이와는 달리 아래와 같이 구주화할 수도 있다.

Job_A_mapper | job_B | job_C_mapper

(Job_A_mapper | job_B_mapper | job_B_reducer | job_C_mapper)

즉, 전처리 단계는 모두 매퍼에서 실행하고, 후처리 단계는 리듀서 이후에 실행하는 방식이다. 이때 ChainMapper와 ChainReducer를 사용할 수 있다. 위의 로직을 의사 코드로 작성해보면 아래와 같다.

ChainMapper.addMapper(Job_A_mapper);

ChainMapper.addMapper(Job_B_mapper);

ChainReducer.setReducer(Job_B_reducer);

ChainReducer.addMapper(Job_C_mapper);

IsolationRunner을 활용한 태스크 재실행

TDD를 바탕으로 MapReduce를 개발하더라도, 실행단계에서는 예외가 발생하기 마련이다. 이러한 경우 스택 트레이스를 눈으로 확인해서 디버깅을 하게되는데, 그 과정이 상당히 번거로울 뿐만 아니라 버그를 찾기도 쉽지 않다. 이를 위해 하둡에서는 ISolationRunner라는 유틸리티를 제공한다. ISolationRunner를 사용하면 실패한 태스크에 대해, 실패할 당시의 데이터를 사용하셔 해당 태스크만 독립적으로 실행할 수 있다.

ISolationRunner를 사용하려면 keep.failed.tasks.files 속성을 true로 설정해야 한다. true로 설정한 후 Job을 실행하면, 태스크가 실패한 경우 해당 데이터를 보관하게 된다.

자세한 사용법은 하둡 튜토리얼을 참고한다.

MapReduce 성능 향상을 위한 팁

Combiner의 활용

매퍼의 결과값이 큰 경우, 매퍼의 결과를 모두 리듀서로 전달할 때 네트워크 I/O에서 병목이 발생할 수 있다. 이러한 경우에는 컴바이너를 활용하면 성능이 향상될 수 있다. 컴바이너란 별도의 프로그램을 구현하기보다는, 리듀서를 매퍼 단계에서 적용하여 매핑의 결과를 줄이는 효과를 내기 위해서 사용한다.

압축하기

컴바이너를 활용하더라도, 매퍼의 결과인 중간 단계의 데이터가 클 수 있다. 이러한 경우 중간 데이터를 압축하면 성능을 개선할 수 있다. 매퍼의 결과를 압축하기 위해서는 아래의 속성을 설정해야 한다.

 

속성	설명
mapred.compress.map.output	매퍼의 출력을 압축하지 여부
mapred.map.output.compression.codec	매퍼의 출력을 압축할 때 사용할 코덱 클래스

 

이 두 속성은 설정파일에 전역 속성으로 설정하기보다는 Job의 성격에 따래 JobConf 객체에 설정하는 편이 낫다. 압축할 때 사용할 수 있는 코덱에는 DefaultCondec, GzipCode, BZip2Code 등이 있다.

 

리듀스의 출력 데이터는 중간단계의 데이터가 아닌 최종 결과물인 경우가 많다. 따라서 리듀스의 출력을 압축하는 경우 문제가 된다. 출력값이 하둡의 블록 사이즈(보통 64MB)보다 큰 경우, 압축된 출력 결과가 2개 이상의 블록으로 분할되게 된다. 대다수의 압축형식에서 이처럼 분할된 압축 파일을 개별적으로 압축 해제할 수 없다. 따라서 리듀서의 출력 값을 입력 값으로 사용하는 또다른 Job이 있는 경우, 각 블록마다 태스크가 실행되는 대신 하나의 태스크만 실행되므로 병렬 처리라는 장점을 잃게 된다.

이를 위해 하둡에서는 SequenceFile이라는 특별한 형태의 파일을 지원한다.

JVM의 재사용

태스크트래커는 매퍼와 리듀서를 별도의 JVM을 새로 할당하여 자식 프로세스로 실행한다. 따라서 초기화 작업이 복잡한 경우, JVM을 실행하는데 걸리는 시간이 오래 걸리게 된다. 이처럼 JVM을 새로 시작하는데 시간이 오래 걸린다면, Job을 실행하는 전체 시간도 오래 걸리게 된다.

하둡 0.19 버전에서는 동일한 Job의 여러 태스크에 대해 JVM을 재사용할 수 있다. JVM을 재사용하면 태스크의 시작 비용을 줄일 수 있게 된다. JVM을 재사용하려면 mapred.job.reuse.jvm.num.tasks 속성을 변경한다. 또는 JobConf 객체의 setNumTasksToExecutePerJvm(int) 메서드를 활용할 수도 있다.

** 주석) 실제로 성능향상이 어느 정도 있는지, 재사용에 따른 side-effect는 없는지는 좀더 확인이 필요할 듯...

추론적 실행(Speculative Execution)

맵리듀스 프레임워크는 맵 태스크나 리듀스 캐스크가 실패한 경우, 자동으로 태스크를 재실행한다. 이처럼 재실행할 수 있는 것은 동일한 태스크가 여러번 실행되더라도 동일한 결과가 나오도록 보장하기 때문이다. 이는 수학적 용어로 멱등성이 보장된다라고 부른다.

태스크가 실패하지 않더라도 동일한 태스크가 여러번 실행될 수 있다. 예를 들어 특정 노드에서 태스크가 느리기 진행될 수 있다. 이는 자원 상황(CPU, I/O 등)에 따라 해당 시점에만 발생할 수 있는 경우다. 이 경우 하둡은 동일한 태스크를 다른 노드에서도 실행하게 되며, 가장 빨리 완료된 태스크의 결과를 최종 결과로 사용하며 종료되지 않은 나머지 태스크는 종료시킨다. 이러한 기능을 추론적 실행이라고 부른다.

기본적으로 추론적 실행은 true로 설정되어 있으며, 맵과 리듀서에 대해 개별적으로 설정할 수 있다.

• mapred.map.tasks.speculative.execution : 매퍼에 대한 설정
• mapred.reduce.tasks.speculative.execution : 리듀서에 대한 설정

태스크가 모두 완료되어야만 Job이 완료되므로, 특정 태스크가 느리게 처리되는 경우 Job의 전체 실행시간 또한 느려지게된다. 따라서 추론적 실행을 사용하면 특정 상황에 종속되지 않고 Job의 실행시간이 평균적으로 빠르게 처리된다는 점을 보장한다.

하지만 일반적으로는 추론적 실행을 사용하지 않는다. 이는 태스크가 멱등성을 가지지 않는 경우가 많기 때문이다. 예를 들어 분석 결과를 DB에 저장하는 경우, 추론적 실행을 사용하면 동일한 데이터가 DB에 저장되므로 중복 저장되는 결과를 낳게 된다. 따라서 실행하려는 Job의 특성에 따라 추론적 실행을 설정해야 한다.

DB와의 상호작용

맵리듀스 프로그램의 입력 데이터가 DB에 있거나, 맵리듀스의 결과를 DB에 저장해야 하는 경우가 있을 수 있다. 이 중에서 입력 데이터를 DB에서 가져와야 하는 경우에는 매퍼에서 DB 커넥션을 얻어서 처리하는 방식은 지양해야 한다. 입력 데이터가 많은 경우, 매퍼의 개수는 그에 맞게 늘어나게 되며 결국 DB 커넥션의 수도 그만큼 증가하게 된다. 이 경우 DB에 과부하가 걸리며 전체 Job의 실행 속도가 늦춰지거나, DB 자체에 장애가 발생할 수도 있다.

따라서 입력 데이터를 DB로부터 읽어야하는 경우에는 드라이버 클래스에서 DB 커넥션을 하나만 얻어와서 파일로 만든 후, 각 맵에서는 해당 파일을 사용하도록 구현해야 한다.

반대로 맵 리듀스의 결과를 DB에 저장해야하는 경우에는 각 리듀서에서 DB 커넥션을 얻어와서 처리할 수 있다. 리듀스는 매퍼에 비해 동시에 실행되는 수가 아주 적으므로, DB 커넥션 또한 많이 사용하지 않기 때문이다. 하둡에서는 이를 위해 DBOutputFormat 클래스를 제공한다.

DBOutputFormat

먼저 드라이버에서 출력 포맷을 DBOutputFormat으로 설정한 후, DB 커넥션 정보는 DBConfiguration 클래스의 configureDB() 메서드를 활용해서 설정한다. 그리고 출력할 데이터를 DBOutputFormat 클래스의 setOutput() 메서드를 활용해서 설정한다.

conf.setOutputFormat(DBOutputFormat.class);

DBConfiguration.configureDB(job, drvierClass, dbUrl, userId, userPass);

DBOutputFormat.setOutput(job, tableName, col1, col2, col3, ...);

DBOutputFormat을 사용할 경우, 오로지 출력 키만에 데이터베이스에 저장된다. 따라서 저장할 데이터에 맞게 출력 키에 대한 클래스를 DBWritable 클래스를 상속받아서 구현해야 한다. 구현 및 활용법은 "Hadoop Map/Reduce 이용하여 HDFS 의 파일 내용을 mySql 에 입력하기"를 참조하라.

태스크 개수 할당 규칙

태스크트래커는 동시에 실행할 수 있는 맵과 리듀서의 최대 개수를 설정할 수 있으며, 기본값은 4개다(맵 2개, 리듀서 2개). 일반적으로 코어 하나당 2개의 태스크를 할당한다. 따라서 쿼드 코어라면 총 6개의 태스크를 설정할 수 있다(맵 3개, 태스크 3개). 태스크 트래커와 데이터노드가 각각 하나의 태스크를 사용하므로, 결국 총 8개의 태스크가 실행되게 된다. 이 경우 전제사항은 해당 태스크에서 CPU를 많이 사용하기 보다는 I/O 작업이 많은 경우다. 만약 CPU 사용이 많은 태스크라면, 설정은 조금 낮추어야 한다.

동시에 실행할 맵과 리듀서의 개수는 각 Job별로 설정할 수도 있다. 맵의 경우 입력 데이터의 크기에 따라 자동으로 설정되며, 리듀서의 개수는 직접 설정이 가능하다. 별도로 설정하지 않는 경우 각 Job별로 하나의 reduce 태스크만 할당된다. 일반적으로 클러스터에서 리듀서 태스크트래커 최대 개수에 0.95 또는 1.75를 곱한 값을 사용한다.

** 주석) 이유는 잘 모르겠다.

fsck

하둡은 파일 시스템의 상태를 검사할 수 있는 fsck 유틸리티를 제공한다.

bin/hadoop fsck /path/to/directory

over-replicated 블록 또는 under-replicated 블록의 경우 시간이 지나면 자동으로 정상으로 돌아온다. 하지만 mis-replicated 블록, corrupt 블록, missing replicas의 경우에는 자동으로 복구가 불가능하다. 이 경우 문제가된 블록을 삭제하려믄 fsck 명령어에 -delete 옵션을 사용한다. 또는 -move 옵션을 사용해 일단 /lost-found로 옮긴 후, 이후에 복구작업을 수행할 수도 있다.

권한 관리

하둡은 일반적으로 하둡이 실행된 시스템의 사용자 권한 체계를 그대로 따른다. 주의할 점은 하둡을 실행한 사용자는 하둡 파일 시스템에서 슈퍼유저로 분류된다. 만약 슈퍼유저 권한을 또다른 사용자에게 주고자한다면, dfs.permissions.supergroup 파라미터를 설정한다.

데이터노드 제거 및 추가

특정 데이터노드가 장애가 난 경우, 단순히 노드를 클러스터에서 네트워크 연결을 끊더라도 HDFS은 정상적으로 운영된다. 그리고 장애가 난 데이터노드에 있던 블록을 복제 계수를 맞추기 위해 자동으로 복제작업을 시작한다. 하지만 좀더 안정적으로 제거하고자 한다면 디커미션(decommsion) 기능을 사용해야 한다.

먼저 네임노드가 설치된 노드에서 dfs.hosts.exclude 파일에 제거할 노드를 "호스트명 IP" 형태로 추가한다. 그런후 네임노드에서 아래의 명령어를 실행한다.

bin/hadoop dfsadmin -refreshNodes

그러면 네임노드는 디커미션 작업을 실행하여 해당 노드를 제거하게 된다.

데이터노드 추가 작업 또한 간단하다. 단순히 데이터노드 데몬과 태스크트래커 데몬을 실행하면 된다. 네임노드가 재실행될 경우를 대비해서 네임노드의 conf/slaves 파일에 추가할 데이터노드를 추가하는 것도 잊지 말자.

이보다 중요한 작업은 데이터를 옮기는 일이다. 추가된 새로운 노드에는 디스크가 비어 있으므로, 기존의 데이터노드에서 블록을 복제하여 새로운 노드에 복사해야 한다. 이를 위해 밸런서를 실행해야 한다.

bin/start-balancer.sh

위의 명령어는 백그라운드에서 실행되며, 데이터 분포가 균형을 이룰때까지 실행된다.

클러스터가 균형을 이룬다는 말은 각 데이터노드의 사용률이 시스템의 평균 사용률의 임계치 내에 들어왔다는 말이다 일반적으로 임계치는 10%이다. 밸런서를 실행할 임계치를 설정할 수 있다. 아래는 임계치가 5%로 설정하여 밸런서를 실행하는 경우다.

bin/start-balancer.sh -threshold 5

네임노드와 세컨데리 네임노드

네임노드는 시스템의 메타데이터와 블록에 대한 위치정보를 메모리에 유지한다. 네임노드는 드라이브 문제로 인한 데이터 손실을 막기 위해서 RAID 드라이브를 사용해야 한다.

네임노드는 HDFS에 저장된 파일의 블록에 대한 매핑정보를 메모리에 유지하므로, 블록이 많아질수록 성능이 느려질 수 있다. 따라서 블록 사이즈를 크게하면 매핑정보가 그만큼 줄어들게 된다. 반면에 블록 사이즈가 커지면, 맵리듀스에서 병렬적으로 처리할 수 있는 태스크의 개수가 작아지므로, 입력 파일의 사이즈가 작은 경우 Job 프로그램의 실행속도가 느려질 수도 있다.

기본적으로 블록의 사이즈는 64MB이며, 경우에 따라 128MB로 설정하기도 한다. 블록 사이즈는 dfs.block.size 속성을 설정하면 된다.

세컨데리 네임노드로 구동할 노드는 반드시 네임노드와 동일한 하드웨어 스펙을 사용해야 한다. 이를 이해하려면 네임노드와 세컨데리 네임노드 사이의 관계에 대해 이해하여야 한다.

네임노드는 fsimage와 edlits 로그 파일을 이용해여 파일 시스템의 상태정보를 관리한다. fsimage 파일은 특정 시점을 파일 시스템 전체에 대한 스냅샷이며, edits 로그 파일은 해당 시점 이후의 변경사항을 기록한 로그 파일이다. 따라서 두 개의 파일을 이용하면 파일 시스템의 현재 상태를 정확히 알 수 있다.

네임노드가 구동될 때 네임노드는 두 파일을 병합하여 현재 시점에 대한 스냡샷을 fsimage 파일에 기록한다. 그리고 비어 있는 edits 로그 파일이 만들어진다. 이후의 변경사항은 모두 edits 로그 파일에 기록되게 된다. 네임노드는 파일 시스템의 상태 정보에 대한 복사본(fsimage와 edits 로그 파일이 병합된 정보)을 메모리에 상주시켜, 클라이언트 요청에 빠르게 응답하게 된다.

이떄 문제는 edits 로그 파일이 너무 커질 경우, 네임노드를 재기동할 때 병합하는 시간이 오래 걸리므로 재기동하는 시간 또한 오래 걸리게 된다. 이를 위해 fsimage 파일과 edits 로그 파일을 병합하는 작업을 주기적으로 실행하게 된다. 네임노드의 경우 파일 시스템 전체에 대한 마스터 역할을 수행하므로, 병합 작업과 같이 자원을 많이 차지하는 작업은 별도의 노드에서 수행하게 되는데, 바로 이 서버가 세컨데리 네임 노드다. 즉 세컨데리 네임노드는 네임 노드로부터 주기적으로 fsimage 파일과 edits 로그 파일을 fs.checkpoint.dir 디렉토리로 가져와서 병합하여, 해당 시점의 스냅샷을 기록한다. 그리고 병합된 fsimage 파일은 다시 네임노드로 복사하여, 네임노드는 병합된 fsimage 파일을 저장하게 된다.

따라서 세컨데리 네임노드는 백업용 노드가 아닌, 현재 상태의 파일 시스템 정보를 기록하기 위한 체크포인트 서버로 이해해야 한다. 결국 세컨데리 네임노드는 네임 노드와 동일한 파일을 관리해야 하므로 최소한 동일한 크기의 메모리를 보장받아야 한다.

네임노드 장애복구

현재 네임노드는 이중화 구성을 할 수 없다. 네임노드는 하둡 파일 시스템에서 단일 장애지점에 해당한다. 일반적으로 세컨데리 네임노드를 백업용 노드로 활용하여, 네임노드에 장애가 발생했을 때를 대비한다. 물론 장애가 발생한다고 해서 세컨데리 네임노드가 자동으로 네임노드의 역할을 수행하는 것은 아니며, 약간의 수고를 들여야한다. 하지만 데이터는 손실되지 않도록 보장할 수 있게 된다.

다수의 디스크를 활용한 장애복구

네임노드에 디스크가 2개 이상인 경우, dfs.name.dir 속성에 해당 디스크를 모두 설정한다. 따라서 첫번째 디스크에 장애가 발생한다면, 네임노드는 두번째로 지정된 디스크를 자동으로 사용하게 된다.

세컨데리 네임노드를 활용한 장애복구

디스카뿐만 아니라 네임노드가 설치된 머신 자체가 장애가 난 경우를 대비해서 세컨데리 네임노드를 백업 노드로 활용할 수 있다. 마찬가지로 dfs.name.dir 속성에 세컨데리 네임노드의 디스크를 추가한다. 그리고 네임노드에 장애가 발생한 경우, 세컨데리 네임노드의 IP의 네임노드의 IP로 변경한 후 네임노드를 재실행한다. 각 데이터 노드는 IP를 기반으로 네임노드와 통신하므로 반드시 IP를 변경해야만 한다.

네임노드 이중화

현재 네임노드를 Active / Stand-by 형태로 이중화할 수 있는 작업이 진행중인 듯하다. 자세한 내용은 아래를 참고하라.

• HDFS High Availability
• Streaming Edits to a Backup Node
  

Job 스케쥴링

기본적으로 하둡은 FIFO 스케쥴러를 사용한다. 즉 모든 Job은 실행된 순서대로 차례차례 실행되게 된다. 따라서 한번에 한개의 Job만을 실행할 수 있다.

하지만 경우에 따라 2개 이상의 Job을 동시에 실행하길 원할 수도 있다. 이를 위해 페이스북에서는 Fair 스케쥴러를, 야후에서는 Capacity 스케쥴러를 구현했다.

Fair 스케쥴러

페어 스케쥴러에서 중요한 개념은 태스크 풀(task pool)이다. 각 Job은 모두 특정 풀에 속하며, 각 풀은 모두 최소한의 맵과 리듀스 개수를 보장받는다. 또한 Job마다 우선순위를 설정할 수 있으며, 우순 선위가 더 높은 Job은 더 많은 태스크 개수를 할당받는다.

페어 스케쥴러를 사용하려면 설정파일을 아래와 같이 수정해야 한다.

<property>

    <name>mapred.jobtracker.taskScheduler</name>

    <value>org.apache.hadoop.mapred.FairScheduler</value>

</property>

<property>

    <name>mapred.fairscheduler.allocation.file</name>

    <value>HADOOP_CONF_DIR/pools.xml</value>

</property>

<property>

    <name>mapred.fairscheduler.assignmultiple</name>

    <value>true</value>

</property>

<property>

    <name>mapred.fairscheduler.poolnameproperty</name>

    <value>pool.name</value>

</property>

<property>

    <name>pool.name</name>

    <value>${user.name}</value>

</property>

위의 설정 파일을 좀더 자세히 살펴보자. 먼저 사용할 스케쥴러의 클래스명을 수정한다.

<property>

    <name>mapred.jobtracker.taskScheduler</name>

    <value>org.apache.hadoop.mapred.FairScheduler</value>

</property>

태스크 풀에 대한 설정정보를 저장한 파일의 경로를 설정한다.

<property>

    <name>mapred.fairscheduler.allocation.file</name>

    <value>HADOOP_CONF_DIR/pools.xml</value>

</property>

Job이 실행될 때 어느 풀을 사용할지 결정하는 기준을 설정한다. 여기에서는 pool.name, 즉 풀의 이름을 기준으로 선택하도록 설정한다.

<property>

    <name>mapred.fairscheduler.poolnameproperty</name>

    <value>pool.name</value>

</property>

pool.name의 기본값을 ${user.name}으로 설정하면, job을 실행한 사용자 계정을 기준으로 사용할 풀을 선택하게 된다.

아래는 풀에 대한 설정을 담고 있는 pools.xml 파일이다.

<?xml version="1.0">

<allocations>

    <pool name="hadoop">

        <minMaps>2</minMaps>

        <minReduces>2</minReduces>

    </pool>

    <pool name="hive">

        <minMaps>2</minMaps>

        <minReduces>2</minReduces>

        <maxRunningJobs>2</maxRunningJobs>

    </pool>
    <user name="bulk">
        <maxRunningJobs>5</maxRunningJobs>
    </user>

    <userMaJobsDefault>3</userMaxJobsDefault>

</allocations>

pools.xml 파일은 매 15초마다 자동으로 다시 읽히므로, 변경사항은 바로바로 반영되게 된다. 이 파일에 정의되지 않은 pool은 최소한의 태스크 수를 보장받지 못하게 된다.

페어 스케쥴러를 사용한 경우 최대 이점은 웹 모니터링 UI 화면이다. http://masternode:50030/scheduler로 접속해보면, 페어스케쥴러에서만 사용할 수 있는 기능을 확인할 수 있다.

 

웹 UI 상에서 풀에 대한 정보 및, Job의 우선순위를 바꿀수도 있다.

 

[참고자료]

거침없이 배우는 하둡 / Chuck Lam 지음 / 이현남, 강택현 옮김 / 지앤선 출판사