[Hadoop] MapReduce 프로그래밍 실습 예제 - Word Count

지난 글에서 Docker기반으로 아주 빠르고 간단하게 Hadoop 실습 환경을 구축해봤다.

이번 글에서는 Hadoop의 핵심인 MapReduce 공식 튜토리얼을 따라 해 보면서, MapReduce 시작의 정석인 WordCount 프로그램을 작성해보고 직접 실행시켜보자.

 

WordCount 프로그램이란? 왜 WordCount인가?

말 그대로, 영단어의 갯수를 세는 프로그램이다. 텍스트 파일에서 각 단어가 몇 개인지를 count 해준다.

예를 들어 아래와 같은 텍스트가 Input으로 들어가면,

Deer Bear River
Car Car River
Deer Car Bear

Output으로 이렇게 각 단어의 개수를 센 결과를 출력해주는 프로그램이다.

Bear	2
Car	3
Deer	2
River	2

 

그렇다면, 왜 WordCount로 MapReduce 설명을 시작하는가?

Hadoop의 핵심은 분산 병렬 처리이다. 일을 한 명이 하는 게 아니라, 여러 명이 똑같은 일을 나눠서 함으로써 효율적으로 처리한다는 것이다. 예를 들어, 위의 샘플처럼 3줄짜리 텍스트의 단어를 센다면 혼자 하면 되겠지만... 책 1,000권의 단어 개수를 센다면? 여러 명이 나눠서 하는 게 좋을 것이다. 이런 의미에서 WordCount는 분산 처리를 설명하기에 아주 적절한 예제이다.

 

0. Hadoop 실행 환경 확인

지난 글을 참고하거나 각자 구축한 환경에서 Hadoop 실행 환경이 잘 갖춰져 있는지 확인하고, 다음 과정부터 쭉 따라 하면 된다.

(Docker 기반이 아니라면 2번으로 바로 넘어가자)

Namenode와 Datanode가 잘 떠있는지 간단하게 확인한다.

docker ps

 

1.  Volume 연결 및 Text 파일 준비

1-1. Volume 확인

레포지토리의 최상단 경로에 docker-compose.yml 파일이 있을 것이다.

여기에서 datanode의 volumes 부분을 확인한다.

Docker에서 Volume을 간단하게 설명하면, 내 컴퓨터(로컬)의 특정 경로와 컨테이너 안에서의 특정 경로를 연동시켜 놓은 것이다.

:을 기준으로 [내 컴퓨터의 경로] : [컨테이너 안에서의 경로]가 docker-compose 파일에 적혀있게 된다.

위 이미지 기준으로 ./data/datanode라는 내 컴퓨터의 경로가 컨테이너 안에서는 /hadoop/dfs/data에 연동되어있다.

 

1-2. Text 파일 준비

내 컴퓨터의 지정 경로로 이동해서, 단어 수를 셀 파일(words.txt를 준비한다)

# 경로 이동
cd ./data/datanode

# 단어를 셀 파일 아무거나 작성하고 :wq로 저장
vim words.txt 

Deer Bear River
Car Car River
Deer Car Bear

 

2. WordCount.java 작성

같은 경로에서 아래와 같이 코드를 작성하고 :wq로 저장한다.

vi WordCount.java

 

WordCount.java 파일 내용은 아래와 같다.(출처: Apache Hadoop 공식 Tutorial )

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.output.FileOutputFormat;

public class WordCount {

  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

 

 

3.  컴파일

3-1.  컨테이너 접속

컨테이너에 접속한 후, 이전에 Volume으로 연동해준 경로로 이동한다. (/hadoop/dfs/data)

방금 로컬에서 생성한 2개의 파일(words.txt, WordCount.java)이 보인다.

# 컨테이너 내부 접속
docker exec -it docker-hadoop-spark-workbench_datanode_1 /bin/bash

# 경로 이동
cd /hadoop/dfs/data

3-2. 환경변수 설정

환경변수 $JAVA_HOME, $HADOOP_CLASSPATH, $PATH 값을 잡아줘야 한다.

# 환경변수 설정
# JAVA_HOME은 되어있는 것을 확인하고, 나머지를 설정해준다.
echo $JAVA_HOME
# /usr/lib/jvm/java-8-openjdk-amd64 가 출력됨

export PATH=${JAVA_HOME}/bin:${PATH}
export HADOOP_CLASSPATH=${JAVA_HOME}/lib/tools.jar

3-3. 컴파일 및 jar 파일 생성

아래 두 개 커맨드를 실행시킨다. 아무 반응이 없으면 정상적으로 실행된 것이다.

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

 

4. 실행

진짜 다했다. 돌려서 확인해보자!

4-1. HDFS에 파일 올리기

아주 간단한 hadoop 커맨드로 로컬 -> HDFS로 파일을 복사해서 올린다.

# HDFS에 올리기
hadoop fs -put ./words.txt /
# 확인
hadoop fs -ls /

 

4-2. jar 파일 실행

java 코드에서 0번째 파라미터로 input 파일의 HDFS 경로를, 1번째 파라미터로 output이 저장될 경로를 명시하게 작성했었다. 

이에 맞춰서 3에서 컴파일하여 생성한 jar파일을 hadoop jar 명령으로 실행시킨다.

hadoop jar wc.jar WordCount /words.txt /output

그러면, 아래와 같이 MapReduce 관련 로그가 찍히면서 MapReduce 프로그램이 실행되는 것을 확인할 수 있다. 

4-3. 확인

실행 시 지정한 Output 경로에 들어가서, 생성된 파일의 내용을 확인해본다.

hadoop fs -ls /output
hadoop fs -cat /output/part-r-00000

아래와 같이 단어의 수가 잘 세어진 것을 확인할 수 있다!

 

끝! 다음 글에서는 WordCount.java 코드를 line-by-line으로 보면서 MapReduce에 대해 이해해볼 예정이다.

 

 

참고

https://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html

Apache Hadoop 3.3.1 – MapReduce Tutorial

Apache Hadoop 3.3.1 – MapReduce Tutorial

지난 글

[Hadoop] Docker 기반 Hadoop 실행 환경 설치/구축