MapReduce 튜토리얼

MapReduce 튜토리얼

이 문서는 Hadoop MapReduce 프레임워크의 모든 사용자 대면 측면을 포괄적으로 설명하고 튜토리얼 역할을 해요.

출처: 문서

본문

목적 (Purpose)

이 문서는 Hadoop MapReduce 프레임워크의 모든 사용자 대면 측면을 포괄적으로 설명하고 튜토리얼 역할을 해요.

전제 조건 (Prerequisites)

Hadoop이 설치·구성·실행 중인지 확인해요. 자세한 내용: 처음 사용자를 위한 Single Node Setup, 대규모 분산 클러스터를 위한 Cluster Setup.

개요 (Overview)

Hadoop MapReduce는 대량의 데이터(멀티-테라바이트 데이터셋)를 상용 하드웨어의 대형 클러스터(수천 노드)에서 신뢰할 수 있고 내결함성 있는 방식으로 병렬 처리하는 애플리케이션을 쉽게 작성하기 위한 소프트웨어 프레임워크예요.

MapReduce 작업은 보통 입력 데이터셋을 독립적인 청크로 나누고, map 태스크가 완전히 병렬로 처리해요. 프레임워크는 map의 출력을 정렬하며, 이것이 그다음 reduce 태스크의 입력이 돼요. 보통 작업의 입력과 출력 모두 파일시스템에 저장돼요. 프레임워크는 태스크 스케줄링, 모니터링, 실패한 태스크의 재실행을 담당해요.

보통 계산 노드와 스토리지 노드는 같아요. 즉 MapReduce 프레임워크와 Hadoop Distributed File System (HDFS Architecture Guide 참고)이 같은 노드 집합에서 실행돼요. 이 구성은 프레임워크가 데이터가 이미 있는 노드에서 태스크를 효과적으로 스케줄링할 수 있게 해주며, 클러스터 전반에 걸쳐 매우 높은 총 대역폭을 만들어내요.

MapReduce 프레임워크는 단일 마스터 ResourceManager, 클러스터-노드당 하나의 워커 NodeManager, 애플리케이션당 하나의 MRAppMaster로 구성돼요 (YARN Architecture Guide 참고). 최소한 애플리케이션은 입출력 위치를 지정하고 적절한 인터페이스 및/또는 추상 클래스의 구현을 통해 map·reduce 함수를 제공해요. 이들과 다른 작업 파라미터가 작업 구성(job configuration)을 구성해요. 그러면 Hadoop 작업 클라이언트는 작업(jar/실행파일 등)과 구성을 ResourceManager에 제출하고, ResourceManager는 소프트웨어/구성을 워커에 배포하고, 태스크를 스케줄링·모니터링하며, 작업-클라이언트에 상태와 진단 정보를 제공하는 책임을 진다.

Hadoop 프레임워크가 Java™로 구현되어 있지만 MapReduce 애플리케이션이 Java로 작성될 필요는 없어요. Hadoop Streaming은 사용자가 어떤 실행파일(예: shell 유틸리티)이든 mapper 및/또는 reducer로 사용해 작업을 만들고 실행할 수 있게 해주는 유틸리티예요. Hadoop Pipes는 C++로 작성된 mapper/reducer와 통신하는 C++ API예요.

입력과 출력 (Inputs and Outputs)

MapReduce 프레임워크는 오직 <key, value> 쌍으로만 동작해요. 즉 프레임워크는 작업에 대한 입력을 <key, value> 쌍 집합으로 보고, 개념적으로 다른 유형일 수 있는 <key, value> 쌍 집합을 작업의 출력으로 생산해요. 키와 값 클래스는 프레임워크가 직렬화할 수 있어야 하므로 Writable 인터페이스를 구현해야 해요. 추가로 키 클래스는 프레임워크가 정렬을 용이하게 하도록 WritableComparable 인터페이스를 구현해야 해요.

MapReduce 작업의 입력·출력 유형:

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

예제: WordCount v1.0 (Example: WordCount v1.0)

자세한 내용을 살펴보기 전에, 어떻게 동작하는지 감을 잡기 위해 예제 MapReduce 애플리케이션을 살펴봐요. WordCount는 주어진 입력 집합에서 각 단어의 발생 횟수를 세는 간단한 애플리케이션이에요. 로컬-스탠드얼론, 의사-분산 또는 완전-분산 Hadoop 설치(Single Node Setup)에서 작동해요.

소스 코드 (Source Code)

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);
  }
}

사용법 (Usage)

환경 변수가 다음과 같이 설정됐다고 가정:

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

WordCount.java를 컴파일하고 jar를 만들어요:

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

가정:

• /user/joe/wordcount/input - HDFS의 입력 디렉터리
• /user/joe/wordcount/output - HDFS의 출력 디렉터리

샘플 텍스트 파일을 입력으로:

$ bin/hadoop fs -ls /user/joe/wordcount/input/
/user/joe/wordcount/input/file01
/user/joe/wordcount/input/file02

$ bin/hadoop fs -cat /user/joe/wordcount/input/file01
Hello World Bye World

$ bin/hadoop fs -cat /user/joe/wordcount/input/file02
Hello Hadoop Goodbye Hadoop

애플리케이션을 실행해요:

$ bin/hadoop jar wc.jar WordCount /user/joe/wordcount/input /user/joe/wordcount/output

출력:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop 2
Hello 2
World 2

애플리케이션은 태스크의 현재 작업 디렉터리에 있을 경로의 쉼표로 구분된 목록을 -files 옵션으로 지정할 수 있어요. -libjars 옵션은 애플리케이션이 map과 reduce의 클래스패스에 jar를 추가할 수 있게 해줘요. -archives 옵션은 쉼표로 구분된 archive 목록을 인수로 전달할 수 있게 해줘요. 이 archive들은 압축이 풀리고 archive 이름의 링크가 태스크의 현재 작업 디렉터리에 만들어져요. 명령줄 옵션에 대한 자세한 내용은 Commands Guide에서 확인할 수 있어요.

-libjars, -files, -archives와 함께 wordcount 예제 실행:

bin/hadoop jar hadoop-mapreduce-examples-<ver>.jar wordcount -files cachefile.txt -libjars mylib.jar -archives myarchive.zip input output

여기서 myarchive.zip은 "myarchive.zip"이라는 이름의 디렉터리에 배치되고 압축이 풀려요. 사용자는 #를 사용해 -files와 -archives 옵션으로 전달된 파일·archive의 다른 기호 이름을 지정할 수 있어요. 예:

bin/hadoop jar hadoop-mapreduce-examples-<ver>.jar wordcount -files dir1/dict.txt#dict1,dir2/dict.txt#dict2 -archives mytar.tgz#tgzdir input output

여기서 dir1/dict.txt와 dir2/dict.txt 파일은 기호 이름 dict1과 dict2를 사용해 태스크가 접근할 수 있어요. archive mytar.tgz는 "tgzdir"라는 이름의 디렉터리에 배치되고 압축이 풀려요.

애플리케이션은 -Dmapreduce.map.env, -Dmapreduce.reduce.env, -Dyarn.app.mapreduce.am.env 옵션을 각각 사용해 mapper, reducer, application master 태스크의 환경 변수를 지정할 수 있어요. 예를 들어 다음은 mapper와 reducer에 대해 환경 변수 FOO_VAR=bar와 LIST_VAR=a,b,c를 설정해요:

bin/hadoop jar hadoop-mapreduce-examples-<ver>.jar wordcount -Dmapreduce.map.env.FOO_VAR=bar -Dmapreduce.map.env.LIST_VAR=a,b,c -Dmapreduce.reduce.env.FOO_VAR=bar -Dmapreduce.reduce.env.LIST_VAR=a,b,c input output

살펴보기 (Walk-through)

WordCount 애플리케이션은 상당히 간단해요.

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);
  }
}

Mapper 구현은 map 메서드를 통해 지정된 TextInputFormat이 제공하는 대로 한 번에 한 줄씩 처리해요. 그런 다음 StringTokenizer를 통해 그 줄을 공백으로 구분된 토큰으로 나누고 < <word>, 1>의 키-값 쌍을 방출해요.

주어진 샘플 입력에 대해 첫 번째 map은 다음을 방출해요:

< Hello, 1>
< World, 1>
< Bye, 1>
< World, 1>

두 번째 map은 다음을 방출해요:

< Hello, 1>
< Hadoop, 1>
< Goodbye, 1>
< Hadoop, 1>

주어진 작업에 대해 생성되는 map 수와 그것을 세밀하게 제어하는 방법은 튜토리얼에서 조금 더 뒤에 배울 거예요.

job.setCombinerClass(IntSumReducer.class);

WordCount는 또한 combiner를 지정해요. 따라서 각 map의 출력이 키로 정렬된 후 로컬 결합(local aggregation)을 위해 로컬 combiner(작업 구성에 따라 Reducer와 같은)를 통과해요.

첫 번째 map의 출력:

< Bye, 1>
< Hello, 1>
< World, 2>

두 번째 map의 출력:

< Goodbye, 1>
< Hadoop, 2>
< Hello, 1>

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);
}

Reducer 구현은 reduce 메서드를 통해 각 키에 대한 발생 횟수(이 예제에서는 단어)인 값들을 합산해요. 따라서 작업의 출력은:

< Bye, 1>
< Goodbye, 1>
< Hadoop, 2>
< Hello, 2>
< World, 2>

main 메서드는 Job에서 입출력 경로(명령줄로 전달), 키/값 유형, 입출력 형식 등과 같은 작업의 다양한 측면을 지정해요. 그런 다음 job.waitForCompletion을 호출해 작업을 제출하고 진행 상황을 모니터링해요. Job, InputFormat, OutputFormat과 다른 인터페이스·클래스에 대해서는 튜토리얼에서 조금 더 뒤에 배울 거예요.

MapReduce - 사용자 인터페이스 (MapReduce - User Interfaces)

이 섹션은 MapReduce 프레임워크의 모든 사용자 대면 측면에 대해 충분한 세부 정보를 제공해요. 이것은 사용자가 작업을 세밀하게 구현·구성·튜닝하는 데 도움이 될 거예요. 그러나 각 클래스/인터페이스의 javadoc이 가장 포괄적인 문서로 남아 있다는 점에 유의하세요. 이것은 단지 튜토리얼로 의도된 것이에요.

먼저 Mapper와 Reducer 인터페이스를 살펴봐요. 애플리케이션은 보통 이들을 구현해 map과 reduce 메서드를 제공해요. 그런 다음 Job, Partitioner, InputFormat, OutputFormat 등을 포함한 다른 핵심 인터페이스를 논의할 거예요. 마지막으로 DistributedCache, IsolationRunner 같은 프레임워크의 유용한 기능 일부를 논의하며 마무리할 거예요.

Payload

애플리케이션은 보통 Mapper와 Reducer 인터페이스를 구현해 map과 reduce 메서드를 제공해요. 이것들이 작업의 핵심을 형성해요.

Mapper

Mapper는 입력 키/값 쌍을 중간 키/값 쌍 집합에 매핑해요. Map은 입력 레코드를 중간 레코드로 변환하는 개별 태스크예요. 변환된 중간 레코드는 입력 레코드와 같은 유형일 필요가 없어요. 주어진 입력 쌍은 0개 또는 많은 출력 쌍에 매핑될 수 있어요.

Hadoop MapReduce 프레임워크는 작업의 InputFormat이 생성한 각 InputSplit에 대해 하나의 map 태스크를 생성해요. 전반적으로 mapper 구현은 Job.setMapperClass(Class) 메서드를 통해 작업에 전달돼요. 그런 다음 프레임워크는 그 태스크의 InputSplit에 있는 각 키/값 쌍에 대해 map(WritableComparable, Writable, Context)을 호출해요. 애플리케이션은 cleanup(Context) 메서드를 재정의해 필요한 정리를 수행할 수 있어요.

출력 쌍은 입력 쌍과 같은 유형일 필요가 없어요. 주어진 입력 쌍은 0개 또는 많은 출력 쌍에 매핑될 수 있어요. 출력 쌍은 context.write(WritableComparable, Writable) 호출로 수집돼요. 애플리케이션은 Counter를 사용해 통계를 보고할 수 있어요. 주어진 출력 키와 연관된 모든 중간 값은 이후에 프레임워크에 의해 그룹화돼요.

Reducer

Reducer는 키와 그 키에 대한 중간 값의 집합을 더 작은 값 집합으로 줄여요. Reduce 태스크는 Reducer가 제공한 reduce 함수로 각 그룹의 중간 값을 줄여 더 작은 값 집합을 생성하는 단계를 수행해요.

주어진 작업에 대해 프레임워크가 생성하는 reduce 태스크 수는 job.setNumReduceTasks(int)으로 지정돼요. 전반적으로 Reducer 구현은 Job.setReducerClass(Class) 메서드를 통해 작업에 전달돼요. shuffle 단계가 완료된 후 프레임워크는 각 그룹에 대해 reduce(WritableComparable, Iterable<Writable>, Context)를 호출해요.

Partitioner

Partitioner는 중간 key-값 쌍을 reduce 태스크에 분배하는 데 사용돼요. 해시 기반 파티셔닝을 수행하는 기본 파티셔너는 reduce 태스크의 수 R에 대해 hash(key) % R을 계산해요. 기본 파티셔너는 org.apache.hadoop.mapreduce.lib.partition.HashPartitioner예요. 애플리케이션은 Job.setPartitionerClass(Class)으로 사용자 정의 파티셔너를 지정할 수 있어요.

Counter

Counter는 통계를 기록하는 도구예요. Mapper, Reducer, 드라이버 코드에서 사용할 수 있어요. 카운터는 조인 등에서 중복을 피하기 위해 유용해요. 자세한 내용은 Counters 섹션을 참고하세요.

Job 구성 (Job Configuration)

작업은 Job 클래스로 구성돼요. Configuration을 받아 작업을 초기화하고, 입출력 경로, map/reduce 클래스, 압축 코덱, 파티셔너, 리듀서 수 등 다양한 작업 속성을 설정할 수 있어요.

태스크 실행과 환경 (Task Execution & Environment)

MRAppMaster가 실행하는 하위 태스크는 JVM 옵션으로 구성할 수 있어요. 예:

<property>
  <name>mapreduce.map.java.opts</name>
  <value>
  -Xmx512M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@[email protected]
  -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false
  </value>
</property>

<property>
  <name>mapreduce.reduce.java.opts</name>
  <value>
  -Xmx1024M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@[email protected]
  -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false
  </value>
</property>

메모리 관리 (Memory Management)

사용자/관리자는 실행되는 하위-태스크와 그것이 재귀적으로 실행하는 어떤 하위 프로세스의 최대 가상 메모리를 mapreduce.{map|reduce}.memory.mb으로 지정할 수 있어요. 여기 설정된 값은 프로세스당 제한이라는 점에 유의하세요. mapreduce.{map|reduce}.memory.mb의 값은 메가바이트(MB)로 지정해야 해요. 그리고 값은 JavaVM에 전달되는 -Xmx보다 크거나 같아야 해요. 그렇지 않으면 VM이 시작하지 않을 수 있어요.

참고: mapreduce.{map|reduce}.java.opts는 MRAppMaster에서 실행되는 하위 태스크를 구성하는 데만 사용돼요. 데몬의 메모리 옵션 구성은 Configuring the Environment of the Hadoop Daemons에 문서화돼 있어요.

프레임워크의 일부 부분에 사용 가능한 메모리도 구성 가능해요. map과 reduce 태스크에서 연산의 동시성과 데이터가 디스크에 닿는 빈도에 영향을 주는 파라미터를 조정하면 성능이 영향을 받을 수 있어요. 작업의 파일시스템 카운터를 모니터링하는 것 — 특히 map에서 나와 reduce로 들어가는 바이트 수에 관해 — 은 이 파라미터의 튜닝에 매우 귀중해요.

Map 파라미터 (Map Parameters)

map에서 방출된 레코드는 버퍼에 직렬화되고 메타데이터는 회계(accounting) 버퍼에 저장돼요. 다음 옵션에 설명된 대로 직렬화 버퍼 또는 메타데이터 중 하나가 임계값을 초과하면, 버퍼의 내용은 정렬되어 백그라운드에서 디스크에 쓰여지며(spill) map은 레코드를 계속 출력해요. 스필(spill)이 진행되는 동안 어느 버퍼든 완전히 차면 map 스레드가 블록돼요. map이 끝나면 남은 레코드는 디스크에 쓰여지고 모든 온-디스크 세그먼트는 단일 파일로 병합돼요.

디스크로의 스필 수를 최소화하면 map 시간을 줄일 수 있지만, 더 큰 버퍼는 mapper에 사용 가능한 메모리를 줄여요.

Name	Type	Description
mapreduce.task.io.sort.mb	int	map에서 방출된 레코드를 저장하는 직렬화·회계 버퍼의 누적 크기 (메가바이트)
mapreduce.map.sort.spill.percent	float	직렬화·회계 버퍼의 소프트 리밋(soft limit) 또는 스필 임계값

셔플/리듀스 파라미터 (Shuffle/Reduce Parameters)

reduce가 시작하기 전에 셔플 단계에서 중간 파일을 가져오고 정렬해요. 셔플/리듀스 성능에 영향을 주는 파라미터: mapreduce.reduce.shuffle.parallelcopies, mapreduce.reduce.shuffle.input.buffer.percent, mapreduce.reduce.shuffle.merge.percent, mapreduce.reduce.merge.inmem.threshold, mapreduce.reduce.input.buffer.percent 등이 있어요.

태스크 로그 (Task Logs)

태스크 로그는 yarn.nodemanager.log-dirs 구성의 노드 로컬 및 로그 디렉터리 계층 구조 아래의 로그 디렉터리에 저장돼요.

라이브러리 배포 (Distributing Libraries)

DistributedCache를 사용해 작업에 필요한 라이브러리와 jar를 배포할 수 있어요. 자세한 내용은 DistributedCache 단락을 참고하세요.

작업 제출과 모니터링 (Job Submission and Monitoring)

Job.submit()으로 작업을 제출하고 Job.waitForCompletion(boolean)으로 완료를 기다릴 수 있어요. Job.getStatus(), Job.getCounters() 등으로 상태를 모니터링할 수 있어요.

InputSplit과 RecordReader (InputSplit, RecordReader)

InputFormat은 데이터를 InputSplit으로 나누고, 각 split은 개별 map 태스크에 할당돼요. RecordReader는 split에서 레코드(키-값 쌍)를 읽어 Mapper에 공급해요.

OutputCommitter (OutputCommitter)

OutputCommitter는 작업의 커밋(commit) 동작을 관리해요 — 작업 초기화 시 임시 출력 디렉터리 생성, 태스크 성공 시 임시 출력을 최종 위치로 승격, 작업 완료 시 정리 등.

RecordWriter (RecordWriter)

RecordWriter는 reduce 출력 키-값 쌍을 출력 파일에 기록해요.

예제: WordCount v2.0 (Example: WordCount v2.0)

WordCount의 두 번째 버전은 MapReduce 프레임워크가 제공하는 일부 기능을 사용해 이전 버전을 개선해요.

소스 코드 (Source Code)

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
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;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.StringUtils;

public class WordCount2 {

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

    static enum CountersEnum { INPUT_WORDS }

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

    private boolean caseSensitive;
    private Set<String> patternsToSkip = new HashSet<String>();

    private Configuration conf;
    private BufferedReader fis;

    @Override
    public void setup(Context context) throws IOException,
        InterruptedException {
      conf = context.getConfiguration();
      caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
      if (conf.getBoolean("wordcount.skip.patterns", false)) {
        URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
        for (URI patternsURI : patternsURIs) {
          Path patternsPath = new Path(patternsURI.getPath());
          String patternsFileName = patternsPath.getName().toString();
          parseSkipFile(patternsFileName);
        }
      }
    }

    private void parseSkipFile(String fileName) {
      try {
        fis = new BufferedReader(new FileReader(fileName));
        String pattern = null;
        while ((pattern = fis.readLine()) != null) {
          patternsToSkip.add(pattern);
        }
      } catch (IOException ioe) {
        System.err.println("Caught exception while parsing the cached file '"
            + StringUtils.stringifyException(ioe));
      }
    }

    @Override
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      String line = (caseSensitive) ?
          value.toString() : value.toString().toLowerCase();
      for (String pattern : patternsToSkip) {
        line = line.replaceAll(pattern, "");
      }
      StringTokenizer itr = new StringTokenizer(line);
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
        Counter counter = context.getCounter(CountersEnum.class.getName(),
            CountersEnum.INPUT_WORDS.toString());
        counter.increment(1);
      }
    }
  }

  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();
    GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    String[] remainingArgs = optionParser.getRemainingArgs();
    if ((remainingArgs.length != 2) && (remainingArgs.length != 4)) {
      System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
      System.exit(2);
    }
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount2.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    List<String> otherArgs = new ArrayList<String>();
    for (int i=0; i < remainingArgs.length; ++i) {
      if ("-skip".equals(remainingArgs[i])) {
        job.addCacheFile(new Path(remainingArgs[++i]).toUri());
        job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
      } else {
        otherArgs.add(remainingArgs[i]);
      }
    }
    FileInputFormat.addInputPath(job, new Path(otherArgs.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1)));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

샘플 실행 (Sample Runs)

샘플 텍스트 파일을 입력으로:

$ bin/hadoop fs -ls /user/joe/wordcount/input/
/user/joe/wordcount/input/file01
/user/joe/wordcount/input/file02

$ bin/hadoop fs -cat /user/joe/wordcount/input/file01
Hello World, Bye World!

$ bin/hadoop fs -cat /user/joe/wordcount/input/file02
Hello Hadoop, Goodbye to hadoop.

애플리케이션을 실행해요:

$ bin/hadoop jar wc.jar WordCount2 /user/joe/wordcount/input /user/joe/wordcount/output

출력:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop, 1
Hello 2
World! 1
World, 1
hadoop. 1
to 1

입력이 우리가 본 첫 번째 버전과 다르고, 그것이 출력에 어떻게 영향을 주는지 주목해요.

이제 DistributedCache를 통해 무시할 단어-패턴을 나열한 패턴-파일을 플러그인해요.

$ bin/hadoop fs -cat /user/joe/wordcount/patterns.txt
\.
\,
\!
to

더 많은 옵션과 함께 다시 실행해요:

$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=true /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

예상대로의 출력:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop 1
Hello 2
World 2
hadoop 1

한 번 더 실행하되, 이번에는 대소문자 구분을 끄고:

$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=false /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

확실히, 출력:

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
bye 1
goodbye 1
hadoop 2
hello 2
world 2

핵심 포인트 (Highlights)

WordCount의 두 번째 버전은 MapReduce 프레임워크가 제공하는 일부 기능을 사용해 이전 버전을 개선해요:

• Mapper(그리고 Reducer) 구현의 setup 메서드에서 애플리케이션이 구성 파라미터에 어떻게 접근할 수 있는지 보여줘요.
• DistributedCache를 사용해 작업에 필요한 읽기 전용 데이터를 어떻게 배포할 수 있는지 보여줘요. 여기서는 사용자가 카운트하는 동안 건너뛸 단어-패턴을 지정할 수 있게 해줘요.
• 제네릭 Hadoop 명령줄 옵션을 처리하는 GenericOptionsParser의 유용성을 보여줘요.
• 애플리케이션이 Counters를 어떻게 사용하고, map(및 reduce) 메서드에 전달되는 애플리케이션 특정 상태 정보를 어떻게 설정할 수 있는지 보여줘요.

더 알아보기 (Learn more)

• 원문: 문서