욱'S 노트

목적

이 문서는 몇개의 노드부턴 수천개의 이르는 노드까지 주요한 하둡 클러스터를 어떻게 설치하고, 설정하고 관리하는지에 대해 설명한다.

먼저 할 일

일단 하둡 릴리이즈 버젼을 다운로드 하자.

설치

하둡 클러스터 설치는 일반적으로 머신에 소프트웨어를 압축을 해제하거나 RPM을 설치하는 것을 얘기한다. 일반적으로 클러스터의 한 대의 머신은 네임노드로 사용되고 나머지 머신은 리소스매니져로 이용된다. 이들이 마스터이다. 나머지 클러스터의 머신들은 데이터노드와 노드매니져이다. 이것이 슬레이브이다.

Non-secure 모드로 하둡 클러스터 기동하기

다음 섹션은 하둡 클러스터를 어떻게 설정하는지에 대해 설명한다. 하둡 설정에 이용되는 두가지 타입의 주요한 설정 파일은 다음과 같가.

• Read-only default configuration - core-default.xml, hdfs-default.xml, yarn-default.xml and mapred-default.xml.
• Site-specific configuration - conf/core-site.xml, conf/hdfs-site.xml, conf/yarn-site.xml and conf/mapred-site.xml.

추가적으로 배포판의 bin/ 디렉토리에서 있는 하둡 스크립트파일(conf/hadoop-env.sh 과 yarn-env.sh)을 수정함으로써 사이트에 맞는 값을 세팅할 수 있다. 하둡 클러스터를 설정하기 위해 하둡 데몬의 환경과 하둡 데몬의 설정 파라미터를 이해할 필요가 있다. 하둡 데몬은 네임노드/데이터노드와 리소스매니져/노드매니져 이다.

하둡 데몬 환경 설정하기

관리자는 conf/hadoop-env.sh과 conf/yarn-env.sh 스크립트를 수정하여 사이트에 맞는 하둡 데몬 프로세스 환경을 커스터마이징할 수 있다. 적어도 각 원격 노드에 JAVA_HOME은 정확하게 지정해야 한다. 많은 경우 하둡 데몬을 수행할려고 하는 유저에 의해 작성될 수 있는 HADDOP_PID_DIR과 HADDOP_SECURE_DN_PID_DIR 디렉토리를 명시해야 한다.  그렇지 않으면 symlink attack의 잠재적인 위험이 발생한다.

관리자는 다음 있는 속성을 설정함으로써 개별적인 데몬 설정을 할 수 있다.

• NameNode - HADOOP_NAMENODE_OPTS
• DataNode - HADOOP_DATANODE_OPTS
• Secondary NameNode - HADOOP_SECONDARYNAMENODE_OPTS
• ResourceManager - YARN_RESOURCEMANAGER_OPTS
• NodeManager - YARN_NODEMANAGER_OPTS
• WebAppProxy - YARN_PROXYSERVER_OPTS
• Map Reduce Job History Server - HADOOP_JOB_HISTORYSERVER_OPTS

  export HADOOP_NAMENODE_OPTS="-XX:+UseParallelGC ${HADOOP_NAMENODE_OPTS}"

다른 유용한 설정 파라미터는 다음과 같다.

HADOOP_LOG_DIR / YARN_LOG_DIR - 데몬의 로그 파일이 저장될 디렉토리. 존재하지 않으면 자동적으로 생성한다.
HADDOP_HEAPSIZE / YARN_HEAPSIZE - 최대 사용할 수 있는 힙사이즈. 만약 1000이라고 세팅하면 1000MB가 할당 된다. 데몬을 위한 힙사이즈 설정시 사용된다. 기본은 1000이다.
ResourceManager	- YARN_RESOURCEMANAGER_HEAPSIZE
NodeManager	- YARN_NODEMANAGER_HEAPSIZE
WebAppProxy	- YARN_PROXYSERVER_HEAPSIZE
Map Reduce Job History Server - HADOOP_JOB_HISTORYSERVER_HEAPSIZE
하둡 데몬  No-Secure모드 설정하기

이번 섹션의 내용은 주어진 설정 파일에 중요한 파라미터들을 설명한다.

conf/core-site.xml

fs.defaultFS - NameNode URI (hdfs://host:port)
io.file.buffer.size - 시퀀스 파일을 쓸때 사용되는 read/write 버퍼 사이즈

conf/hdfs-site.xml

네임노드 설정

dfs.namenode.name.dir - 네임노드가 네임스페이스와 트랜잭션로그를 저장할 로컬 파일시스템 경로. 콤마로 분리된 디렉토리리스트면 redundancy를 위해 네임테이블은 복제된다.
dfs.namenode.hosts / dfs.namenode.hosts.exclude - 포함 또는 제외될  데이터노드 리스트 (필요하다면 접근 데이터노드 리스트를 관리하기 위해 사용할 수 있다.)
dfs.blocksize - 대용량 파일 시스템을 위한 HDFS 블럭사이즈 (256MB)
dfs.namenode.handler.count - 많은 수의 데이터노드로 부터 RPC를 처리할 네임 노드 스레드 수 (100)

데이터노드 설정

dfs.datanode.data.dir - 콤마로 분리된 데이터노드가 블럭을 저장할 로컬 파일 시스템 리스트 (redundancy를 위해 복제될 것이다.)

conf/yarn-site.xml
 
리소스매니져 및 노드매니져 설정
yarn.acl.enable - acl 활셩화 여부. 기본값은 false
yarn.admin.acl - Admin ACL. ALC은 콤마로 분리된 유저 스페이스 콤마로 분리된 그룹. 기본값은 *은 모두를 의미하고 단순한 스페이스는 아무도 접근할 수 없다는 것을 의미한다.
yarn.log-aggregation-enable - 로그 조합을 가능하게 할지 설정. 기본값은 false.

리소스매니져 설정
yarn.resourcemanager.address - 작업 요청을 위한 클라이언트를 위한 리소스매니져 host:port. 세팅하면 yarn.resourcemanager.hostname의 호스트명이 오버라이드된다.
yarn.resourcemanager.scheduler.address - 어플리케이션마스터들와  스케쥴러간의 요청을 주고 받기 위한 리소스매니져 host:port. 만약 세팅하면 yarn.resourcemanager.hostname의 호스트명이 오버라이드된다.
yarn.resourcemanager.resource-tracker.address - 노드매니져를 위한 리소스매니져 host:port. 세팅하면 yarn.resourcemanager.hostname의 호스트명이 오버라이드된다.
yarn.resourcemanager.admin.address - 어드민 커맨드를 위한 리소스매니져 host:port. 세팅하면 yarn.resourcemanager.hostname의 호스트명이 오버라이드된다.
yarn.resourcemanager.webapp.address - 리소스매니져 웹UI host:port. 세팅하면 yarn.resourcemanager.hostname의 호스트명이 오버라이드된다.
yarn.resourcemanager.hostname - 리소스매니져 호스트. 하나의 호스트명은 yarn.resourcemanager.*.address의 호스트명으로 세팅된다. 결과적으로 포트는 리소스매니져 컴포넌트의 기본값을 사용하게 된다.
yarn.resourcemanager.scheduler.class - 리소스매니져 스케쥴러 클래스. CapacityScheduler(추천), FairScheduler(추천) 또는 FifoScheduler
yarn.scheduler.minimum-allocation-mb - 리소스 매니져의 각 컨테이너 요청에 할당될 최소 메모리
yarn.scheduler.maximum-allocation-mb - 리소스 매니져의 각 컨테이너 요청에 할당될 최대 메모리
yarn.resourcemanager.nodes.include-path/yarn.resourcemanager.nodes.exclude-path - 노드매니져 포함 또는 제외 리스트. (필요하다면 접근 노드매니져 리스트를 관리하기 위해 사용할 수 있다.)
노드매니져 설정
yarn.nodemanager.resource.memory-mb - 노드매니져에게 주어진 이용가능한 물리 메모리.  노드매니져에 총 이용가능한 리소스는 실행중인 컨테이너를 사용될 수 있게 정의되어야 한다.
yarn.nodemanager.vmem-pmem-ratio - 물리 메모리를 초과하였을 경우 작업이 사용할 가상 메모리의 비율. 각 작업의 메모리 사용량은 물리 메모리 사용량을 비율에 따라 초과할 수 있다. 
yarn.nodemanager.local-dirs - 콤마로 분리된 중간데이터를 작성할 로컬 파일시스템의 경로 리스트. 멀티 패스는 디스크 I/O를 분산한다.
yarn.nodemanager.log-dirs - 콤마로 분리된 로그를 작성할 로컬 파일시스템의 경로 리스트, 멀티 패스는 디스크 I/O를 분산한다.
yarn.nodemanager.log.retain-seconds - 노드매니져의 로그 파일을 획득하기 위한 기본 시간. log-aggregation이 비활성화 되었을 경우에만 적용된다.
yarn.nodemanager.remote-app-log-dir - 어플리케이션 수행완료시 어플리케이션 로그를 옮길 HDFS 디렉토리, 적당한 권한이 필요하고 log-aggregation이 활성화 되었을 경우에만 적용된다. (logs)
yarn.nodemanager.remote-app-log-dir-suffix - 리모트 로그 디렉토리에 붙는 suffix. 로그는 ${yarn.nodemanager.remote-app-log-dir}/${user}/{$thisParam}. log-aggregation이 활성화 되었을 경우에만 적용된다. (logs)
yarn.nodemanager.aux-services - 맵리듀스 어플리케이션을 위해 세팅될 필요가 있는 shuffle 서비스 (mapreduce_shuffle)

히스토리 서버 설정
yarn.log-aggregation.retain-seconds - 어플리케이션 aggregation 로그를 유지하는 시간. 너무 작게 설정하면 네임노드에 스팸이 될 수 있다. (기본값 : -1)
yarn.log-aggregation.retain-check-interval-seconds - aggregation 로그 유지를 위해 체크하는 시간 인터벌. 0 이나 음수로 세팅하면 0.1초로 계산할 것이다

conf/mapred-site.xml

맵리듀스 어플리케이션 설정
mapreduce.framework.name - Hadoop YARN을 위한 실행 프레임워크(yarn)
mapreduce.map.memory.mb - 맵을 위해 사용될 리소스 용량(1536)
mapreduce.map.java.opts - 맵의 자식 jvm의 힙사이즈(-Xmx1024m)
mapreduce.reduce.memory.mb - 리듀스를 위해 사용될 리소스 용량(3072)
mapreduce.reduce.java.opts - 리듀스의 자식 jvm의 힙사이즈 (-Xmx2560m)
mapreduce.task.io.sort.mb - 데이터 소팅을 위해 사용될 메모리 사이즈 (512)
mapreduce.task.io.sort.factor - 파일 소팅중 한번에 머지할 수 있는 스트림(100)
mapreduce.reduce.shuffle.parallelcopies - 맵의 데이터를 패치하기 위해 리듀스에 의해 생성될 병렬 복사본 수 

맵리듀스 잡히스토리 서버 설정
mapreduce.jobhistory.address - 맵리듀스 작업히스토리 서버 host:port(디폴트 - 10020)
mapreduce.jobhistory.webapp.address - 맵리듀스 작업히스터리 서버 웹 UI host:port(디폴트 : 19888)
mapreduce.jobhistory.inrermediate-done-dir - 맵리듀스 작업으로 작성된 히스트로 파일이 위치할 디렉토리 (/mr-history/tmp)
mapreduce.jobhistory.done-dir - 맵리듀스 작업히스토리 서버에 의해 관리될 히스토리 파일 디렉토리 (/mr-history/done)
Hadoop Rack Awareness

HDFS와 YARN의 컴포넌트들은 rack-aware이다. 네임노드와 리소스매니져는 어드민 설정 모듈에 할당된 API를 호출하기 위해 클러스터 슬레이브의 랙정보를 획득한다. API는 DNS명 또는 ip에 대한 rack id 정보를 유지한다. 사이트 특화된 모듈은 topology.node.switch.mapping.impl 아이템 설정을 활용해 설정될 수 있다. 기본 구현은 topology.script.file.name을 사용해 설정된 script/command를 실행 할 수 있다. 만약 topology.script.file.name이 세팅되어 있지 않다면 모든 전달된 IP address에 기본값을 반환할 것이다

Monitoring Health of NodeManagers

하둡은 관리자로 부터 노드의 헬스체크를 위해 주기적으로 어드민이 제공한 스크립트를 실행하기 위해 노드매니져를 설정할 수 있는 매커니즘을 제공한다. 관리자는 스크립트를 통해 체크를 수행함으로써 노드의 정상여부를 판단할 수 있다. 만약 스크립트가 노드가 죽었다고 판단하면 ERROR와 함께 표준출력으로  추력해야한다. 노드매니져는 주기적으로 스크립트를 실행하고 그것을 결과를 체크한다. 만약 스크립트의 출력이 ERROR를 포함하고 있다면 노드의 상태는 비정상으로 리포트되고 리소스매니져에 의해 노드는 블랙리스트화 된다. 그리고 노드에는 더 이상 작업이 할당되지 않을 것이다. 하지만 노드매니져가 스크립트 수행을 계속하면 노드가 정상으로 판별될 수 있다. 그럴 경우 리소스매니져는 자동으로 블랙리스트에서 제거할 것이다. 노드의 정상여부에 따라 만약 비정상이라면 리소스매니져 웹인터페이스에서 관리자는 조회할 수있다. 노드가 언제 부터 정상인지 또한 웹 인터페이스에 디스플레이에 나타난다. 다음은 노드 정상여부 모니터링 스크립트의 설정을 조정하는 방법인다.

conf/yarn-site.xml

yarn.nodemanager.health-checker.script.path - 노드 헬스상태를 체크하는 스크립트
yarn.nodemanager.health-checker.script.opts - 노드 헬스상태 스크립트를 위한 옵션
yarn.nodemanager.health-checker.script.interval-ms - 헬스 스크립트 수행을 위한 타임 인터벌
yarn.nodemanager.health-checker.script.timeout-ms - 헬스 스크립트 수행을 위한 타임아웃

헬스 체크 스크립트는 로컬 디스크 문제와 같은 에러면 수행될 수 없다. 노드매니져는 로컬 디스크의 정상여부를 주기적으로 체크하고 불량 디렉토리수가 yarn.nodemanager.disk-health-check.min-healthy-disk 에 설정된 임계점에 도달 하면 비정상으로 마크하고 해당 정보를 리소스매니져로 전송한다. 헬스 체크 스크립트에 의해 부트 디스크가 레이드 되어 있는지 실패인지를 판별한다.

Slave files

일반적으로 네임노드 전용으로 동작할 하나의 머신을 선택하고 리소스매니져 전용으로 동작할 하나의 머신을 선택한다. 그리고 나머지를 데이터노드 또는 노드매니져 즉 슬레이브로 사용한다. 모든 슬레이브의 호스트명 혹은 아이피 주소는 conf/slaves  파일에 명시되어야 한다.

Logging

하둡은 로깅을 위해 log4j를 사용한다. conf/log4j.properties를 수정하므로써 하둡 데몬의 로깅 설정을 커스터마이징할 수 있다.

Hadoop Cluster 오퍼레이팅

Hadoop Startup

하둡 클러스터를 시작하기 위해 HDFS와 YARN 클러스터를 모두 시작해야 한다.

새로운 분산 파일시스템 포맷하기:
$ $HADOOP_PREFIX/bin/hdfs namenode -format <cluster_name>

네임노드로 지정된 HDFS 시작하기
$ $HADOOP_PREFIX/sbin/hadoop-daemon.sh --config $HADOOP_CONF_DIR --script hdfs start namenode

모든 슬레이브에 있는 데이터노드 시작하기 
$ $HADOOP_PREFIX/sbin/hadoop-daemon.sh --config $HADOOP_CONF_DIR --script hdfs start datanode

리소스 매니져로 지정된 YARN 시작하기
$ $HADOOP_YARN_HOME/sbin/yarn-daemon.sh --config $HADOOP_CONF_DIR start resourcemanager

모든 슬레이브에 있는 노드매니져 시작하기
$ $HADOOP_YARN_HOME/sbin/yarn-daemon.sh --config $HADOOP_CONF_DIR start nodemanager

WebAppProxy서버 시작하기. 로드밸런싱을 위해 여러대를 구동할 경우 각각 수행한다.
$ $HADOOP_YARN_HOME/sbin/yarn-daemon.sh start proxyserver --config $HADOOP_CONF_DIR

맵리듀스 작업히스토리 서버 구동하기 
$ $HADOOP_PREFIX/sbin/mr-jobhistory-daemon.sh start historyserver --config $HADOOP_CONF_DIR

Hadoop Shutdown

네임노드 중지
$ $HADOOP_PREFIX/sbin/hadoop-daemon.sh --config $HADOOP_CONF_DIR --script hdfs stop namenode

데이터노드 중지
$ $HADOOP_PREFIX/sbin/hadoop-daemon.sh --config $HADOOP_CONF_DIR --script hdfs stop datanode

리소스매니져 중지
$ $HADOOP_YARN_HOME/sbin/yarn-daemon.sh --config $HADOOP_CONF_DIR stop resourcemanager

노드매니져 중지
$ $HADOOP_YARN_HOME/sbin/yarn-daemon.sh --config $HADOOP_CONF_DIR stop nodemanager

WebaAppProxy서버 중지
$ $HADOOP_YARN_HOME/sbin/yarn-daemon.sh stop proxyserver --config $HADOOP_CONF_DIR

맵리듀스 작업히스토리 서버 중지
$ $HADOOP_PREFIX/sbin/mr-jobhistory-daemon.sh stop historyserver --config $HADOOP_CONF_DIR

자 이제 MapReduce로 첫번째 프로그램을 만들어보자. 우리의 목표는 공식사이트의 첫번째 예제인 WordCount이다. 무작정 베끼기만 하면 이해가 어려울 수 있으니 목표는 똑같으나 스타일은 내 스타일대로 가겠다. 목표는 Hadoop의 README.txt 파일의 단어 갯수를 세는 프로그램이다. 

먼저 MapReduce의 mapper 역할을 하는 클래스를 만들어 보자.

import java.io.IOException;
import java.util.StringTokenizer;
 
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 
public class SimpleWordMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final Logger logger = LoggerFactory.getLogger(SimpleWordMapper.class);
    
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        logger.info("Key : {} , Value : {}", key, value);
        
        StringTokenizer tokenizer = new StringTokenizer(value.toString());
        
        while (tokenizer.hasMoreTokens()) {
            String token = tokenizer.nextToken();
            
            context.write(new Text(token), new IntWritable(1));
        }
    }
 
}
 

Mapper 클래스를 보면 generic으로 4개의 타입을 명시하도록 요구한다. 각 타입은 <입력키, 입력값, 출력키, 출력값>에 대한 타입이다. 개요에서 봤듯이 자체 직렬화 및 정렬을 위해 Hadoop에서는 특정 타입을 구현하도록 유도한다. 위의 LongWritable, IntWritable, Text는 쉽게 보면 long, int, String에 대응되는 타입들이라고 보면 되겠다. 해당 클래스는 쉽게 보면 특정 long과 String을 입력으로 받아서 그것의 출력으로 단어 하나와 숫자 1을 context에 쓰는 일을 한다. 보충하자면 입력키와 값이 long과 String인 건 일반 텍스트 파일을 입력으로 활용할 경우 해당 타입의 값으로 입력이 생성된다. 

다음은 reduce 역할을 수행하는 클래스이다. 

import java.io.IOException;
 
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 
public class SimpleWordReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private final Logger logger = LoggerFactory.getLogger(SimpleWordReducer.class);
    
    @Override
    public void reduce(Text key, Iterable<IntWritable> value, Context context) throws IOException, InterruptedException {
        int sum = 0;
        
        for (IntWritable intWritable : value) {
            logger.info("Key : {} , Value : {}", key, intWritable);
            
            sum += intWritable.get();
        }
        
        context.write(key, new IntWritable(sum));
    }
}
 

전에 말했다시피 reducer의 입력은 mapper의 출력이다. 그러나 특이한 점은 value가 iterator형태로 전달되고 있음을 알 수 있다. 먼가 당연하지 않겠는가? 그렇지 않다면 mapper의 출력 결과를 unique하게 만들어야 하는데, 그렇다면 너무 힘들뿐더라 할 수 있는 일이 너무 제약이 될 것이다.

로컬에서 수행을 할려면 hadoop-site.xml 파일을 아래와 같이 수정하면 된다. 클러스터에 작업을 의뢰 할려면  fs.default.name만 수정하면 된다.

 
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
 
<configuration>
    <property>
        <name>fs.default.name</name>
        <value>file:///</value>
    </property>
    <property>
        <name>mapred.job.tracker</name>
        <value>local</value>
    </property>
</configuration>
 

구동하는 소스이다.

 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 
public class SimpleWordCounter {
    private final static String CONFIG_PATH = "/Users/devsun/dev/workspace/test/sample-project/src/test/resources/hadoop-local.xml";
 
    private final static Logger logger = LoggerFactory.getLogger(SimpleWordCounter.class);
    
    private final static String SOURCE_FILE_PATH = "/Users/devsun/dev/workspace/test/sample-project/src/test/resources/README.txt";
    private final static String TARGET_FILE_PATH = "/Users/devsun/dev/workspace/test/sample-project/src/test/resources/output";
 
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.addResource(new Path(CONFIG_PATH));
 
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(SimpleWordCounter.class);
        job.setMapperClass(SimpleWordMapper.class);
        job.setCombinerClass(SimpleWordReducer.class);
        job.setReducerClass(SimpleWordReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        Path sourcePath = new Path(SOURCE_FILE_PATH);
        
        logger.info(sourcePath.getFileSystem(job.getConfiguration()).getHomeDirectory().toString());
                
        FileInputFormat.addInputPath(job, sourcePath);
        FileOutputFormat.setOutputPath(job, new Path(TARGET_FILE_PATH));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
 

작업을 하나 정의하게 되고, 파일의 이력과 출력을 명시한다. 최종 결과를 확인해보자. 역시 잘된다.

다음으로 살펴볼 내용은 MapReduce 이다. 이 넘이 아마 분산처리를 주도하는 기술이 아닐까 싶다.

1. 개요 

그럼, MapReduce란 무엇인가? 

MapReduce는 대량의 클러스터에서 대량의 데이터를 병렬로 처리하는 어플리케이션을 쉽게 만들 수 있도록 도와주는 소프트웨어 프레임워크이다.

공식사이트의 정의를 빌어보자면, 아마 이 넘이 병렬 및 대량처리에 핵심인듯 하다. 대충 돌아가는 개요를 보면 아래와 같다.

MapReduce 작업은 일반적으로 입력데이터를 map task의 처리에 의해 독립적인 chunk들로 분할을 한다. 프레임워크는 해당 map의 출력의 정렬된 경과를 reduce task의 입력으로 전달한다. 통상적으로 작업의 입력과 출력 모두 파일시스템에 저장되고, 프레임워크는 task의 스케쥴링, 모니터링 및 실패된 task의 재실행을 담당한다.

통상적으로 계산하는 노드와 저장하는 노드는 같다. MapReduce 프레임워크와 HDFS는 같은 노드 집합에서 수행된다.이러한 설정은 프레임워크가 데이터의 현재와 결과 클러스터에 매우 강하게 결합되어 task를 효과적으로 스케쥴할 수 있다.

MapReduce 프레임워크는 전체에 하나의 master ResourceManager, 클러스터 노드당 하나의 slave NodeManager 그리고 어플리케이션당 MRAppMaster로 구성된다.

최소한 어플리케이션은 입력/출력 위치를 명시해야하며, 적절한 map과 reduce 기능을 제공해야 한다.

Hadoop 작업 클라이언트는 ResourceManager에 작업과 작업에 대한 설정을 제출하고, Resourcemanger는 slave들로 소프트웨어와 설정을 분배하고, ResourceManager는 task를 스케쥴링 및 모니터링을 하고 상태와 진단 정보를 작업클라언트에 제공한다.

2. 입력과 출력

MapReduce 프레임워크는 상화배타적인 <key, value> 쌍으로 동작한다. 즉, 프레임워크는 작업의 입력된 <key, value> 쌍을 다른 타입의 <key, value> 쌍으로 출력한다.

Key와 Value의 클래스들은 프레임워크에 의해 직렬화되어야 하므로 Writable 인터페이스를 구현해야 한다. 또한 Key 클래스들은 프레임워크에 의해 정렬이 되어야 하므로, WritableComparable 인터페이스를 구현하여야 한다.

(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)

먼말이지 모르겠다. 역시 소스를 봐야 겠다.

HDFS를 다루기 위해서 파일시스템을 추상화한 클래스인 FileSystem 클래스를 제공한다.

Outline을 보면 알겠지만, 파일처리에 대한 모든 api를 제공한다.

일단 간단하게 디렉토리를 한번 만들어보자. 메소드가 명시적이라 어려운이 없을것이라고 생각된다.

 
import java.io.IOException;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 
public class SimpleFileSytemTest {
    private static final String CONFIG_PATH = "/Users/devsun/dev/hadoop-2.5.2/etc/hadoop/core-site.xml";
    private static final String FILE_PATH = "/user/devsun/test/README.txt";
    
    private final Logger logger = LoggerFactory.getLogger(SimpleFileSytemTest.class);
    
    private Configuration configuration;
    
    @Before
    public void setup() {
        configuration = new Configuration();
        configuration.addResource(new Path(CONFIG_PATH));
    }
    
    @Test
    public void testMakeDir() throws IllegalArgumentException, IOException {
        boolean success = FileSystem.get(configuration).mkdirs(new Path("/user/devsun/test2"));
        Assert.assertEquals(true, success);
    }
}
 

다음 소스는 파일을 읽는 소스이다. 일단 파일을 하나 올려보자.

점점 이 정도는 식은죽먹기가 되어간다. 아래는 파일을 읽는 소스이다.

 
    @Test
    public void testFileRead() throws IOException {
        FileSystem fileSystem = null;
        FSDataInputStream inputStream = null;
        StringBuilder contents = new StringBuilder();
        
        try {
            fileSystem = FileSystem.get(configuration);
            
            Path path = new Path(FILE_PATH);
            
            inputStream = fileSystem.open(path);
                    
            byte[] bytes = new byte[4096];
            
            while ( inputStream.read(bytes) != -1) {
                contents.append(new String(bytes));
            }
            
            logger.info("Contents : {}",contents.toString());
            
        } catch (IOException e) {
            logger.error("Error occured.", e);
            throw e;
        } finally {
            inputStream.close();
            fileSystem.close();
        }        
    }
 

우리는 개발자 아닌가?

HDFS를 커맨드로만 조작한다면 그건 TA일 것이다. 그렇다 HDFS를 조작하는 자바 프로그램을 작성해보도록 하자. 일단 먼가 원격의 HDFS를 접속할려면 클라이언트에 설정이 있어야 할 것 같다.

역시 Configuration 객체를 제공하고 있었다. 클라이언트 설정을 위해서 우리가 해야 할일은 굉장히 간단하다.

우리는 기존에 core-site.xml에  HDFS의 접근 URL을 설정하였다. 해당 URL을 프로퍼티처럼 세팅하고 접속을 시도해보면 된다. 아래의 예제는 프로그래밍으로 HDFS에 접속하여 해당 home 디렉토리를 출력한 예제이다.

 
    @Test
    public void testProgrammaticConfiguration() throws IOException {
        Configuration configuration = new Configuration();
        configuration.set("fs.default.name", "hdfs://localhost:9000");
        
        FileSystem dfs = null;
        
        try {
            dfs = FileSystem.get(configuration);
            logger.info("HDFS home directory is {}",dfs.getHomeDirectory());
        } finally {
            
            if (dfs != null) {
                dfs.close();
            }
        }
    }
 

역시 잘된다. 엄청쉽다. 근데 설정이 코드에 들어가는건 별로 좋지 않지 않은가? 다음은 해당 설정을 파일에 수정하고 접속을 수행해보자. 일단 core-site.xml을 좀 수정해보자.

<configuration>
        <property>
                <name>fs.default.name</name>
                <value>hdfs://localhost:9000</value>
        </property>
</configuration>
 

 
@Test
    public void testFileConfiguration() throws IOException {
        Configuration configuration = new Configuration();
        configuration.addResource(new Path("/Users/devsun/dev/hadoop-2.5.2/etc/hadoop/core-site.xml"));
        
        FileSystem dfs = null;
        
        try {
            dfs = FileSystem.get(configuration);
            logger.info("HDFS home directory is {}",dfs.getHomeDirectory());
        } finally {
            
            if (dfs != null) {
                dfs.close();
            }
        }
    }
 

자 이제 어딘가에 원격 파일시스템에 파일도 올리고 카피도 하고 이름도 바꾸고를 할 수 있게 되었다. 그러면 도대체 이때까지 봤던 단어들이 무엇인지를 알아 보자.

1. NameNode, DataNode

HDFS는 master-worker 패턴으로 구성되어 있다. 여기서 master의 역할을 수행하는 것이 NameNode이다. NameNode는 파일시스템의 논리적인 메타정보들을 관리한다. 파일 시스템의 namespace, tree 그리고 그 외 모든 파일들의 메타데이터를 유지한다.

DataNodes는 파일시스템의 worker로서 외부의 요청에 따라 실질적으로 블록들을 저장하고 탐색하고, 해당 내용을 NameNode에 알리는 역할을 수행한다.

다음으로 무엇을 해볼까 생각해본다. 일단 HDFS를 다뤄보자.

파일시스템이라고 했으니 파일 카피하고 디렉토리 조회는 할 줄 알아야 겠지?

일단, HDFS를 구동해본다.

그리고 접속해본다. http://localhost:50070/

근데 뭥미? 브라우져로 확인해보니 404가 뜬다. 이런 말도 안되는...

내렸다 올리면 파일이 날아가나? 먼가 이유가 있겠지? 검색을 해보자.

역시 누군가 경험한 내용이었다. 감사합니다. 

Mac을 껐다 켜보겠다. tmp 아래의 폴더가 날아가나. 놀랍다. 방금 껐다키니 날아간다. tmp란 디렉토리명 답다.

namenode 포맷을 하면 /tmp 폴더에 hadoop namenode가 생성이 되는데 이 폴더는 서버를 내렸다가 올리면 사라진다. 그러므로 서버를 shutdown시키면 안되는 것이다. 

이 namenode 명령어는 namenode를 Running 시키는 명령어인데, start-all.sh나 start-dfs.sh를 실행하면 자동으로 함께 실행된다. 전에 서버를 내렸다가 올렸을때 namenode가 실행되지 않은 이유는 namenode 폴더가 없었기 때문이였다. 

출처 : http://naleejang.tistory.com/31

그러면 다른 디렉토리로 지정하는 방법을 찾아봐야 겠다.

찾았다. $HADOOP_PREFIX/etc/common/hdfs-site.xml 을 수정하자. 네임노드와 데이터노드의 경로를 아래와 같이 지정한다. 그리고 또 껐다 켜봐도 괜찮다.

일반적인 명령어들의 테스트를 해보자. 생각보다 너무 쉽다. skip하겠다. 아래와 같은 쉘들을 제공한다. 아래와 같은 형태로 사용하면 되며, 자세한 사용법은 공식 홈페이지를 참조하면 되겠다.

$ bin/hdfs dfs <args>

• appendToFile
• cat
• chgrp
• chmod
• chown
• copyFromLocal
• copyToLocal
• count
• cp
• du
• dus
• expunge
• get
• getfacl
• getfattr
• getmerge
• ls
• lsr
• mkdir
• moveFromLocal
• moveToLocal
• mv
• put
• rm
• rmr
• setfacl
• setfattr
• setrep
• stat
• tail
• test
• text
• touchz

어제는 어떻게든 하둡을 설치하고, 샘플을 돌려보았다. 이쯤에서 드는 의문이 있다.

그럼 하둡은 과연 무엇인가? 해당 내용은 그냥 하둡 공식 홈페이지에서 발췌했다.

The Apache™ Hadoop® reliable, scalable, distributed computing을 위해 오픈소스 소프트웨어이다. 

The Apache Hadoop software library는 단순한 프로그래밍 모델을 이용하여 클러스터링을 활용하여 large data sets의 분산처리를 도와주는 프레임워크이다. single 환경으로부터 수천대의 machine로 확장이 용이하게 설계되었다. 하드웨어에 의존하는 high-availability와는 달리,어플리케이션 레이어에서 실패를 감지하고 처리한다.

먼가 좋은거 같다.

Hadoop은 크게 네가지 파트로 구성된다.

Hadoop Common: 다른 Hadoop 모듈을 지원하기 위한 공통 유틸리티

Hadoop Distributed File System (HDFS): 고성능의 어플리케이션 데이터를 접근을 제공하는 분산 파일 시스템.

Hadoop YARN: 작업을 스케쥴링하고 클러스터 자원관리를 위한 프레임워크.

Hadoop MapReduce: 대량 데이터 병렬 처리를 담당하는 YARN 기반의 시스템.

오늘부터 hadoop에 대한 공부를 시작할려고 한다.

사실 필자는 Mac도 첨 쓰는지라 엄청 버벅되더라도 정리를 하고 가려고 한다.

1. Hadoop 구하기 및 설치  

Hadoop도 JVM위에서 돌아가는지라 설치는 정말 간단하다. 공식 사이트에 가서 받고 일단 풀기만 하면 된다.

공식사이트는 아시다시피 http://hadoop.apache.org/ 이다.

적당한 최신 버젼을 받고 필자는 2.6.0이 최신이었으나 마이너버젼이 너무 안올라간 관계로 2.5.2를 받았다.

그러면 당연히 다운로드 폴더에 아래와 같은 파일이 다운로드 된다.

그러면 원하는 위치에 풀면 된다.

우리가 항상 설치하고 해보는거 아래와 같이 쳐보자. 대충 버젼이랑 잘 나온다.

./hadoop version  

2. Hadoop Cluster 준비하기

아직 하둡의 개념이 없지 않은가? 그러니 시키는 대로 해보자.

Hadoop 설치한 디렉토리로 가보면 etc/hadoop/hadoop-env.sh 라는 파일이 있다.

이름만 봐도 이 쉘은 hadoop 기본 설정 쉘같이 생겼다.

아래와 같이 환경변수 두개를 export 하자

다들 영어가 되시겠지만, 자바가 설치된 디렉토리랑 하둡이 설치된 디렉토리를 지정하면 된다. HADOOP_HOME이 아닌 이유는 멀까?

# set to the root of your Java installation
export JAVA_HOME=/usr/java/latest

# Assuming your installation directory is /usr/local/hadoop
export HADOOP_PREFIX=/usr/local/hadoop

근데 Mac OS에서 자바 홈은 좀 지랄같다. 그래서 필자는 구글링 결과 다음과 같이 JAVA_HOME을 지정하였다.

export JAVA_HOME=`/usr/libexec/java_home -v 1.8`

3. SSH 설정하기 

로컬이지만 하둡에 원격으로 접속하기 위해 SSH 설정을 해야 한다고 한다.
한번 해보자. 일단 원격 접속을 허용한다.

시스템 환경설정 > 공유로 가보자. 아래와 같이 허용하자.





그리고 터미널에서 접속해보자.



접속이 잘되는 것을 확인할 수 있다.


4. Pseudo-Distributed 모드로 테스트 하기

공식 사이트를 참조해보면 Hadoop은 아래와 같이 세가지 모드를 지원한다고 한다. 로컬모드야 먼가 너무 간단한거 같고 로컬은 아마 가상분산모드 정도가 적합할 것 같아서 가상분산모드로 테스트를 수행해보겠다.
Local (Standalone) Mode
Pseudo-Distributed Mode
Fully-Distributed Mode

해당 파일 두개의 설정을 아래와 같이 변경하자. 오옷! 먼가 저 로컬호스트로 붙을 수 있을 것 같다.

etc/hadoop/core-site.xml

<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value>
    </property>
</configuration>

etc/hadoop/hdfs-site.xml:

<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>

다음은 수행법이라고 한다. 아무것도 모를때는 무작정 따라하자.

아래대로 따라하면 맵리듀스 작업을 로컬에서 수행하는 것이라고 한다.

파일시스템을 포맷한다.

  $ bin/hdfs namenode -format

네임노드 데몬과 데이터노드 데몬을 시작한다. 

  $ sbin/start-dfs.sh

아래 주소를 접속해보자. 놀랍다... 접속이 된다.

http://localhost:50070/

디렉토리를 만들자.

  $ bin/hdfs dfs -mkdir /user

  $ bin/hdfs dfs -mkdir /user/<username>

입력파일을 분산 파일 시스템에 카피하자.

  $ bin/hdfs dfs -put etc/hadoop input

몇개의 제공된 예제를 수행해보자.

  $ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.5.2.jar grep input output 'dfs[a-z.]+'

결과도 확인해보자. 브라우져에서 바로 결과를 다운로드 할 수 있다. 신기하다.

Utilities > Browse the file system으로 가면 된다.

수고했다. 분산 파일 시스템을 종료하자.

  $ sbin/stop-dfs.sh