[Hadoop] 하둡 MapReduce 기본 실습 가이드 (Fully Distributed 모드)

Hadoop MapReduce실습 - WordCount

 

standalone 모드로 실습하는 법 (HDFS 사용X)

https://konkukcodekat.tistory.com/251

[Hadoop] 하둡 설치 및 Standalone 모드로 wordCount 기본 예제 실습

 

Fully Distributed 모드로 실행하기 위해서는 HDFS 실행이 돼야한다.

https://konkukcodekat.tistory.com/252

[Hadoop] 하둡 HDFS 실습 환경 설정 (Fully Distributed 모드)

 

Hadoop이란?

• 대용량 데이터를 신뢰성 있게, 확장 가능하게, 분산 처리할 수 있도록 설계된 오픈소스 프레임워크이다.
• 간단한 프로그래밍 모델로 수천 대의 서버에 분산 저장 및 처리가 가능하다.

Hadoop 구성 모듈

1. HDFS (Hadoop Distributed File System)
   • 분산 저장을 담당
   • NameNode: 메타데이터 관리 (파일의 위치, 구조)
   • DataNode: 실제 데이터 블록 저장 및 제공
   • Secondary NameNode: NameNode의 주기적 백업
2. YARN (Yet Another Resource Negotiator)
   • 자원 관리 및 작업 스케줄링 담당
   • ResourceManager: 전체 자원 할당
   • NodeManager: 노드 단위 자원 모니터링
3. MapReduce
   • 병렬 데이터 처리 프레임워크
   • Map 단계: 데이터를 key-value 형태로 변환
   • Reduce 단계: 중간 결과를 취합 및 집계
4. Hadoop Common
   • 하둡 전반에 필요한 라이브러리와 유틸리티 제공

---

HDFS 동작 원리

Hadoop Block

• 파일을 기본 128MB 블록 단위로 분할 저장
• 블록 단위 분산 저장으로 병렬 처리 성능 향상

Block Replication

• 각 블록은 설정된 복제 계수(기본 3)에 따라 여러 DataNode에 저장됨
• 일부 노드 장애 시에도 데이터 접근 가능 → Fault Tolerance 확보

---

Hadoop MapReduce 기본 구조

1. Input: HDFS에서 데이터를 로드
2. Map: 데이터를 변환하여 (Key, Value) 생성 (병렬)
3. Shuffle & Sort: 중간 결과를 키 기준으로 정렬하고 그룹화
4. Reduce: 키별 값 리스트를 받아 집계 및 최종 결과 생성
5. Output: 결과를 다시 HDFS에 저장

예시:

• (Map) (가,1), (나,1), (가,1) →
• (Shuffle & Sort) (가, [1,1]), (나, [1]) →
• (Reduce) (가, 2), (나, 1)

---

Fully Distributed Mode 실습 가이드

Hadoop 클러스터 시작

start-dfs.sh   # NameNode, DataNode 실행
start-yarn.sh  # ResourceManager, NodeManager 실행

상태 확인

jps  # 모든 노드에서 실행 중인 데몬 확인

---

WordCount 실습

1. HDFS에 input 폴더 생성

hdfs dfs -mkdir /input

2. 로컬에 입력 파일 생성

mkdir ~/input
cd ~/input
touch test1.txt test2.txt
# test1.txt, test2.txt에 원하는 단어들 입력

3. 파일을 HDFS로 업로드

hdfs dfs -put ~/input/test1.txt /input/
hdfs dfs -put ~/input/test2.txt /input/

---

사용자 WordCount 코드 실행

1. WordCount 자바 코드 작성

cd ~/hadoop
vi MyWordCount.java

 

MyWordCount.java

import java.io.IOException;
import java.util.StringTokenizer;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class MyWordCount {

    // Mapper 클래스 정의
    public static class MyMapper extends Mapper<Object, Text, Text, IntWritable> {

        // static final 상수로 정의하여, 공유
        // 1을 값으로 설정 → 어떤 단어가 1번 나올 때 (단어, 1)로 설정하기 위함
        private final static IntWritable one = new IntWritable(1);

        // 단어를 담는 용도 → 미리 생성하여 성능 향상
        // 매번 객체 생성하면 비효율적이므로 재사용
        private Text word = new Text();

        // map 함수 → 한 줄씩 처리
        public void map(Object key, Text value, Context context) 
                throws IOException, InterruptedException {

            // text value : 한 줄 단위로 들어오는 함수의 입력
            // Context : 결과를 Hadoop으로 전달하기 위한 연결고리
            String line = value.toString();  // Hadoop의 Text → Java String 변환

            // 특수기호 제거
            String match = "[^\\uAC00-\\uD7A3xfea-zA-Z\\s]";
            line = line.replaceAll(match, "");

            // 공백을 기준으로 분리된 토큰들이 st 변수에 저장
            // ex) line = "hello hadoop word" → st = ["hello", "hadoop", "word"]
            StringTokenizer st = new StringTokenizer(line);

            // 토큰을 하나씩 꺼내서 처리
            while (st.hasMoreTokens()) {
                // 하나씩 담음
                // nextToken() : 다음 단어 꺼냄
                // toLowerCase() : 모두 소문자로 변환
                word.set(st.nextToken().toLowerCase());

                // context.write : ('단어', 1) 형태로 출력
                context.write(word, one);
            }
        }
    }

    // Reducer 클래스 정의
    public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        // Key로 들어온 입력들의 개수를 합하기 위한 변수
        private IntWritable sumWritable = new IntWritable();

        // reduce 함수 → 같은 Key(단어)에 대해 그룹된 값들을 처리
        protected void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {

            // sum : 단어의 등장 횟수를 누적할 변수
            int sum = 0;

            // 반복문 돌면서 IntWritable 하나씩 가져와 누적
            for (IntWritable val : values) {
                sum += val.get();
            }

            // sumWritable에 단어의 총 개수 저장
            sumWritable.set(sum);

            // context.write : (단어, 총 개수)로 출력
            context.write(key, sumWritable);
        }
    }

    // main 함수 → Job 설정
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "WordCount"); // Job 이름 설정

        job.setJarByClass(MyWordCount.class);        // Jar 클래스 설정
        job.setMapperClass(MyMapper.class);          // Mapper 클래스 지정
        job.setReducerClass(MyReducer.class);        // Reducer 클래스 지정

        job.setOutputKeyClass(Text.class);           // 출력 Key 자료형
        job.setOutputValueClass(IntWritable.class);  // 출력 Value 자료형

        job.setInputFormatClass(TextInputFormat.class);   // 입력 포맷
        job.setOutputFormatClass(TextOutputFormat.class); // 출력 포맷

        // 입력/출력 경로 지정
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // Job 실행 후 종료
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

2. 컴파일 및 JAR 파일 생성

hadoop com.sun.tools.javac.Main MyWordCount.java
jar cf MyWordCount.jar MyWordCount*.class

3. 실행

hadoop jar MyWordCount.jar MyWordCount /input /output

---

결과 확인 및 종료

결과 확인

hdfs dfs -ls /output
hdfs dfs -cat /output/*

클러스터 종료

stop-dfs.sh
stop-yarn.sh

---

기타 설정 파일 참고

mapred-site.xml 에러 발생 시

에러 메시지에 따라 mapred-site.xml을 편집해 작업 모드를 명시해준다.

 

예:

<property>
  <name>mapreduce.framework.name</name>
  <value>yarn</value>
</property>

---

본 실습은 로컬 VM 환경에서 구성된 Fully Distributed Hadoop Cluster에서 진행된다.

테스트용 입력 파일을 로컬에서 생성한 뒤 HDFS로 업로드하고, 직접 구현한 MapReduce 프로그램을 실행하여 분산처리 결과를 확인할 수 있다.