MapReduce-퀵 가이드

MapReduce는 여러 노드에서 병렬로 빅 데이터를 처리 할 수있는 애플리케이션을 작성하기위한 프로그래밍 모델입니다. MapReduce는 방대한 양의 복잡한 데이터를 분석하기위한 분석 기능을 제공합니다.

빅 데이터 란?

빅 데이터는 기존 컴퓨팅 기술을 사용하여 처리 할 수없는 대규모 데이터 세트의 모음입니다. 예를 들어 Facebook 또는 Youtube에서 매일 수집하고 관리하는 데 필요한 데이터의 양은 빅 데이터 범주에 속할 수 있습니다. 그러나 빅 데이터는 규모와 양에 관한 것이 아니라 속도, 다양성, 양 및 복잡성과 같은 측면 중 하나 이상을 포함합니다.

왜 MapReduce인가?

기존 엔터프라이즈 시스템에는 일반적으로 데이터를 저장하고 처리하는 중앙 집중식 서버가 있습니다. 다음 그림은 기존 엔터프라이즈 시스템의 개략도를 보여줍니다. 기존 모델은 확장 가능한 대량의 데이터를 처리하는 데 적합하지 않으며 표준 데이터베이스 서버에서 수용 할 수 없습니다. 더욱이 중앙 집중식 시스템은 여러 파일을 동시에 처리하는 동안 너무 많은 병목 현상을 만듭니다.

Google은 MapReduce라는 알고리즘을 사용하여이 병목 문제를 해결했습니다. MapReduce는 작업을 작은 부분으로 나누어 여러 컴퓨터에 할당합니다. 나중에 결과는 한곳에서 수집되고 통합되어 결과 데이터 세트를 형성합니다.

MapReduce는 어떻게 작동합니까?

MapReduce 알고리즘에는 Map 및 Reduce라는 두 가지 중요한 작업이 포함됩니다.

  • Map 태스크는 데이터 세트를 가져 와서 개별 요소가 튜플 (키-값 쌍)로 분할되는 다른 데이터 세트로 변환합니다.

  • 축소 작업은 맵의 출력을 입력으로 가져와 해당 데이터 튜플 (키-값 쌍)을 더 작은 튜플 집합으로 결합합니다.

축소 작업은 항상 맵 작업 후에 수행됩니다.

이제 각 단계를 자세히 살펴보고 그 중요성을 이해해 보겠습니다.

  • Input Phase − 여기에는 입력 파일의 각 레코드를 번역하고 구문 분석 된 데이터를 키-값 쌍의 형태로 매퍼에게 보내는 레코드 리더가 있습니다.

  • Map − Map은 일련의 키-값 쌍을 가져와 각각을 처리하여 0 개 이상의 키-값 쌍을 생성하는 사용자 정의 함수입니다.

  • Intermediate Keys − 매퍼에 의해 생성 된 키-값 쌍을 중간 키라고합니다.

  • Combiner− 결합기는 맵 단계의 유사한 데이터를 식별 가능한 집합으로 그룹화하는 로컬 감속기 유형입니다. 매퍼에서 중간 키를 입력으로 사용하고 사용자 정의 코드를 적용하여 한 매퍼의 작은 범위에서 값을 집계합니다. 이것은 주요 MapReduce 알고리즘의 일부가 아닙니다. 선택 사항입니다.

  • Shuffle and Sort− Reducer 작업은 Shuffle 및 Sort 단계로 시작됩니다. 그룹화 된 키-값 쌍을 Reducer가 실행중인 로컬 머신에 다운로드합니다. 개별 키-값 쌍은 키별로 더 큰 데이터 목록으로 정렬됩니다. 데이터 목록은 동일한 키를 그룹화하여 Reducer 작업에서 해당 값을 쉽게 반복 할 수 있도록합니다.

  • Reducer− Reducer는 그룹화 된 키-값 쌍 데이터를 입력으로 취하고 각각에 대해 Reducer 기능을 실행합니다. 여기에서 데이터는 여러 가지 방법으로 집계, 필터링 및 결합 될 수 있으며 광범위한 처리가 필요합니다. 실행이 끝나면 최종 단계에 0 개 이상의 키-값 쌍을 제공합니다.

  • Output Phase − 출력 단계에는 Reducer 함수에서 최종 키-값 쌍을 변환하고 레코드 작성기를 사용하여 파일에 쓰는 출력 포맷터가 있습니다.

작은 다이어그램의 도움으로 Map & f Reduce의 두 가지 작업을 이해해 보겠습니다.

MapReduce- 예제

MapReduce의 힘을 이해하기 위해 실제 사례를 살펴 보겠습니다. Twitter는 하루에 약 5 억 개의 트윗을 수신하며 이는 초당 거의 3000 개의 트윗입니다. 다음 그림은 Tweeter가 MapReduce를 사용하여 트윗을 관리하는 방법을 보여줍니다.

그림과 같이 MapReduce 알고리즘은 다음 작업을 수행합니다.

  • Tokenize − 트윗을 토큰 맵으로 토큰 화하고 키-값 쌍으로 작성합니다.

  • Filter − 토큰 맵에서 원하지 않는 단어를 필터링하고 필터링 된 맵을 키-값 쌍으로 작성합니다.

  • Count − 단어 당 토큰 카운터를 생성합니다.

  • Aggregate Counters − 유사한 카운터 값의 집계를 관리 가능한 작은 단위로 준비합니다.

MapReduce 알고리즘에는 Map 및 Reduce라는 두 가지 중요한 작업이 포함됩니다.

  • 맵 작업은 Mapper 클래스를 통해 수행됩니다.
  • 감소 작업은 Reducer Class를 통해 수행됩니다.

매퍼 클래스는 입력을 받아 토큰 화하고 매핑하고 정렬합니다. Mapper 클래스의 출력은 Reducer 클래스의 입력으로 사용되며 일치하는 쌍을 검색하여 줄입니다.

MapReduce는 다양한 수학적 알고리즘을 구현하여 작업을 작은 부분으로 나누고 여러 시스템에 할당합니다. 기술적 인 측면에서 MapReduce 알고리즘은 Map & Reduce 작업을 클러스터의 적절한 서버로 보내는 데 도움이됩니다.

이러한 수학적 알고리즘에는 다음이 포함될 수 있습니다.

  • Sorting
  • Searching
  • Indexing
  • TF-IDF

정렬

정렬은 데이터를 처리하고 분석하는 기본 MapReduce 알고리즘 중 하나입니다. MapReduce는 매퍼의 출력 키-값 쌍을 키별로 자동 정렬하는 정렬 알고리즘을 구현합니다.

  • 정렬 방법은 매퍼 클래스 자체에서 구현됩니다.

  • 셔플 및 정렬 단계에서 매퍼 클래스의 값을 토큰 화 한 후 Context class (사용자 정의 클래스)는 일치하는 값 키를 컬렉션으로 수집합니다.

  • 유사한 키-값 쌍 (중간 키)을 수집하기 위해 Mapper 클래스는 RawComparator 키-값 쌍을 정렬하는 클래스입니다.

  • 주어진 Reducer에 대한 중간 키-값 쌍 세트는 Reducer에 표시되기 전에 키-값 (K2, {V2, V2,…})을 형성하기 위해 Hadoop에 의해 자동으로 정렬됩니다.

수색

검색은 MapReduce 알고리즘에서 중요한 역할을합니다. 결합기 단계 (선택 사항)와 감속기 단계에서 도움이됩니다. 예를 들어 검색이 어떻게 작동하는지 이해해 보겠습니다.

예

다음 예제는 MapReduce가 검색 알고리즘을 사용하여 주어진 직원 데이터 세트에서 가장 높은 급여를받는 직원의 세부 정보를 찾는 방법을 보여줍니다.

  • 직원 데이터가 A, B, C, D의 네 가지 파일에 있다고 가정하겠습니다. 또한 모든 데이터베이스 테이블에서 직원 데이터를 반복적으로 가져 오기 때문에 4 개의 파일 모두에 중복 된 직원 레코드가 있다고 가정하겠습니다. 다음 그림을 참조하십시오.

  • The Map phase각 입력 파일을 처리하고 직원 데이터를 키-값 쌍으로 제공합니다 (<k, v> : <emp 이름, 급여>). 다음 그림을 참조하십시오.

  • The combiner phase(검색 기술)은 직원 이름과 급여가 포함 된 키-값 쌍으로 맵 단계의 입력을 받아들입니다. 컴 바이 너는 검색 기법을 사용하여 모든 직원 급여를 확인하여 각 파일에서 가장 높은 급여를받는 직원을 찾습니다. 다음 스 니펫을 참조하십시오.

<k: employee name, v: salary>
Max= the salary of an first employee. Treated as max salary

if(v(second employee).salary > Max){
   Max = v(salary);
}

else{
   Continue checking;
}

예상 결과는 다음과 같습니다.

<satish, 26000>

<gopal, 50000>

<키란, 45000>

<마니 샤, 45000>

  • Reducer phase− 각 파일을 작성하면 급여가 가장 높은 직원을 찾을 수 있습니다. 중복을 방지하려면 모든 <k, v> 쌍을 확인하고 중복 항목이있는 경우 제거하십시오. 4 개의 입력 파일에서 오는 4 개의 <k, v> 쌍 사이에 동일한 알고리즘이 사용됩니다. 최종 출력은 다음과 같아야합니다.

<gopal, 50000>

인덱싱

일반적으로 인덱싱은 특정 데이터와 해당 주소를 가리키는 데 사용됩니다. 특정 매퍼의 입력 파일에 대해 일괄 인덱싱을 수행합니다.

MapReduce에서 일반적으로 사용되는 인덱싱 기술은 다음과 같이 알려져 있습니다. inverted index.Google 및 Bing과 같은 검색 엔진은 반전 된 인덱싱 기술을 사용합니다. 간단한 예제를 통해 인덱싱이 작동하는 방식을 이해해 보겠습니다.

예

다음 텍스트는 반전 된 인덱싱을위한 입력입니다. 여기서 T [0], T [1] 및 t [2]는 파일 이름이고 해당 내용은 큰 따옴표로 묶여 있습니다.

T[0] = "it is what it is"
T[1] = "what is it"
T[2] = "it is a banana"

인덱싱 알고리즘을 적용한 후 다음과 같은 출력을 얻습니다.

"a": {2}
"banana": {2}
"is": {0, 1, 2}
"it": {0, 1, 2}
"what": {0, 1}

여기서 "a": {2}는 T [2] 파일에 "a"라는 용어가 표시됨을 의미합니다. 마찬가지로 "is": {0, 1, 2}는 "is"라는 용어가 T [0], T [1] 및 T [2] 파일에 표시됨을 의미합니다.

TF-IDF

TF-IDF는 Term Frequency-Inverse Document Frequency의 줄임말 인 텍스트 처리 알고리즘입니다. 일반적인 웹 분석 알고리즘 중 하나입니다. 여기서 '빈도'라는 용어는 문서에서 용어가 나타나는 횟수를 의미합니다.

기간 빈도 (TF)

문서에서 특정 용어가 발생하는 빈도를 측정합니다. 문서에 단어가 나타나는 횟수를 해당 문서의 총 단어 수로 나눈 값입니다.

TF(the) = (Number of times term the ‘the’ appears in a document) / (Total number of terms in the document)

역 문서 빈도 (IDF)

용어의 중요성을 측정합니다. 텍스트 데이터베이스의 문서 수를 특정 용어가 나타나는 문서 수로 나눈 값으로 계산됩니다.

TF를 계산하는 동안 모든 용어는 똑같이 중요한 것으로 간주됩니다. 즉, TF는 "is", "a", "what"등과 같은 일반 단어에 대한 용어 빈도를 계산합니다. 따라서 다음을 계산하여 희귀 단어를 확장하면서 빈번한 용어를 알아야합니다.

IDF(the) = log_e(Total number of documents / Number of documents with term ‘the’ in it).

알고리즘은 작은 예제의 도움으로 아래에 설명되어 있습니다.

예

1000 개의 단어가 포함 된 문서를 생각해보십시오. hive50 번 나타납니다. TFhive 그러면 (50/1000) = 0.05입니다.

이제 1,000 만 개의 문서와 hive이 중 1000 개에 나타납니다. 그런 다음 IDF는 log (10,000,000 / 1,000) = 4로 계산됩니다.

TF-IDF 무게는 0.05 × 4 = 0.20의 양의 곱입니다.

MapReduce는 Linux 기반 운영 체제에서만 작동하며 Hadoop 프레임 워크가 내장되어 있습니다. Hadoop 프레임 워크를 설치하려면 다음 단계를 수행해야합니다.

JAVA 설치 확인

Hadoop을 설치하기 전에 시스템에 Java를 설치해야합니다. 다음 명령을 사용하여 시스템에 Java가 설치되어 있는지 확인하십시오.

$ java –version

Java가 이미 시스템에 설치되어있는 경우 다음 응답이 표시됩니다.

java version "1.7.0_71"
Java(TM) SE Runtime Environment (build 1.7.0_71-b13)
Java HotSpot(TM) Client VM (build 25.0-b02, mixed mode)

시스템에 Java가 설치되어 있지 않은 경우 아래 단계를 따르십시오.

자바 설치

1 단계

다음 링크 ( 이 링크) 에서 최신 버전의 Java를 다운로드하십시오 .

다운로드 후 파일을 찾을 수 있습니다. jdk-7u71-linux-x64.tar.gz 다운로드 폴더에 있습니다.

2 단계

다음 명령을 사용하여 jdk-7u71-linux-x64.gz의 내용을 추출하십시오.

$ cd Downloads/
$ ls jdk-7u71-linux-x64.gz $ tar zxf jdk-7u71-linux-x64.gz
$ ls
jdk1.7.0_71 jdk-7u71-linux-x64.gz

3 단계

모든 사용자가 Java를 사용할 수 있도록하려면 "/ usr / local /"위치로 이동해야합니다. 루트로 이동하여 다음 명령을 입력하십시오-

$ su
password:
# mv jdk1.7.0_71 /usr/local/java
# exit

4 단계

PATH 및 JAVA_HOME 변수를 설정하려면 ~ / .bashrc 파일에 다음 명령을 추가하십시오.

export JAVA_HOME=/usr/local/java
export PATH=$PATH:$JAVA_HOME/bin

현재 실행중인 시스템에 모든 변경 사항을 적용합니다.

$ source ~/.bashrc

5 단계

다음 명령을 사용하여 Java 대안을 구성하십시오-

# alternatives --install /usr/bin/java java usr/local/java/bin/java 2

# alternatives --install /usr/bin/javac javac usr/local/java/bin/javac 2

# alternatives --install /usr/bin/jar jar usr/local/java/bin/jar 2

# alternatives --set java usr/local/java/bin/java

# alternatives --set javac usr/local/java/bin/javac

# alternatives --set jar usr/local/java/bin/jar

이제 명령을 사용하여 설치를 확인하십시오. java -version 터미널에서.

Hadoop 설치 확인

MapReduce를 설치하기 전에 시스템에 Hadoop이 설치되어 있어야합니다. 다음 명령을 사용하여 Hadoop 설치를 확인하겠습니다.

$ hadoop version

Hadoop이 이미 시스템에 설치되어 있으면 다음과 같은 응답을 받게됩니다.

Hadoop 2.4.1
--
Subversion https://svn.apache.org/repos/asf/hadoop/common -r 1529768
Compiled by hortonmu on 2013-10-07T06:28Z
Compiled with protoc 2.5.0
From source with checksum 79e53ce7994d1628b240f09af91e1af4

시스템에 Hadoop이 설치되지 않은 경우 다음 단계를 진행하십시오.

Hadoop 다운로드

Apache Software Foundation에서 Hadoop 2.4.1을 다운로드하고 다음 명령을 사용하여 콘텐츠를 추출합니다.

$ su
password:
# cd /usr/local
# wget http://apache.claz.org/hadoop/common/hadoop-2.4.1/
hadoop-2.4.1.tar.gz
# tar xzf hadoop-2.4.1.tar.gz
# mv hadoop-2.4.1/* to hadoop/
# exit

의사 분산 모드에서 Hadoop 설치

다음 단계는 의사 분산 모드에서 Hadoop 2.4.1을 설치하는 데 사용됩니다.

1 단계-Hadoop 설정

~ / .bashrc 파일에 다음 명령을 추가하여 Hadoop 환경 변수를 설정할 수 있습니다.

export HADOOP_HOME=/usr/local/hadoop
export HADOOP_MAPRED_HOME=$HADOOP_HOME
export HADOOP_COMMON_HOME=$HADOOP_HOME export HADOOP_HDFS_HOME=$HADOOP_HOME
export YARN_HOME=$HADOOP_HOME export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export PATH=$PATH:$HADOOP_HOME/sbin:$HADOOP_HOME/bin

현재 실행중인 시스템에 모든 변경 사항을 적용합니다.

$ source ~/.bashrc

2 단계-Hadoop 구성

"$ HADOOP_HOME / etc / hadoop"위치에서 모든 Hadoop 구성 파일을 찾을 수 있습니다. Hadoop 인프라에 따라 해당 구성 파일을 적절하게 변경해야합니다.

$ cd $HADOOP_HOME/etc/hadoop

Java를 사용하여 Hadoop 프로그램을 개발하려면 다음에서 Java 환경 변수를 재설정해야합니다. hadoop-env.sh JAVA_HOME 값을 시스템의 Java 위치로 대체하여 파일.

export JAVA_HOME=/usr/local/java

Hadoop을 구성하려면 다음 파일을 편집해야합니다.

  • core-site.xml
  • hdfs-site.xml
  • yarn-site.xml
  • mapred-site.xml

core-site.xml

core-site.xml에는 다음 정보가 포함되어 있습니다.

  • Hadoop 인스턴스에 사용되는 포트 번호
  • 파일 시스템에 할당 된 메모리
  • 데이터 저장을위한 메모리 제한
  • 읽기 / 쓰기 버퍼 크기

core-site.xml을 열고 <configuration> 및 </ configuration> 태그 사이에 다음 속성을 추가합니다.

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

hdfs-site.xml

hdfs-site.xml은 다음 정보를 포함합니다-

  • 복제 데이터의 가치
  • 네임 노드 경로
  • 로컬 파일 시스템의 데이터 노드 경로 (Hadoop 인프라를 저장하려는 위치)

다음 데이터를 가정 해 보겠습니다.

dfs.replication (data replication value) = 1

(In the following path /hadoop/ is the user name.
hadoopinfra/hdfs/namenode is the directory created by hdfs file system.)
namenode path = //home/hadoop/hadoopinfra/hdfs/namenode

(hadoopinfra/hdfs/datanode is the directory created by hdfs file system.)
datanode path = //home/hadoop/hadoopinfra/hdfs/datanode

이 파일을 열고 <configuration>, </ configuration> 태그 사이에 다음 속성을 추가합니다.

<configuration>

   <property>
      <name>dfs.replication</name>
      <value>1</value>
   </property>
   
   <property>
      <name>dfs.name.dir</name>
      <value>file:///home/hadoop/hadoopinfra/hdfs/namenode</value>
   </property>
   
   <property>
      <name>dfs.data.dir</name>
      <value>file:///home/hadoop/hadoopinfra/hdfs/datanode </value>
   </property>
   
</configuration>

Note − 위 파일에서 모든 속성 값은 사용자 정의되며 Hadoop 인프라에 따라 변경할 수 있습니다.

yarn-site.xml

이 파일은 yarn을 Hadoop으로 구성하는 데 사용됩니다. yarn-site.xml 파일을 열고 <configuration>, </ configuration> 태그 사이에 다음 특성을 추가하십시오.

<configuration>
   <property>
      <name>yarn.nodemanager.aux-services</name>
      <value>mapreduce_shuffle</value>
   </property>
</configuration>

mapred-site.xml

이 파일은 우리가 사용하는 MapReduce 프레임 워크를 지정하는 데 사용됩니다. 기본적으로 Hadoop에는 yarn-site.xml의 템플릿이 포함되어 있습니다. 먼저 다음 명령을 사용하여 mapred-site.xml.template에서 mapred-site.xml 파일로 파일을 복사해야합니다.

$ cp mapred-site.xml.template mapred-site.xml

mapred-site.xml 파일을 열고 <configuration>, </ configuration> 태그 사이에 다음 속성을 추가합니다.

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

Hadoop 설치 확인

다음 단계는 Hadoop 설치를 확인하는 데 사용됩니다.

1 단계-이름 노드 설정

다음과 같이 "hdfs namenode -format"명령을 사용하여 namenode를 설정하십시오.

$ cd ~ $ hdfs namenode -format

예상 결과는 다음과 같습니다.

10/24/14 21:30:55 INFO namenode.NameNode: STARTUP_MSG:
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG: host = localhost/192.168.1.11
STARTUP_MSG: args = [-format]
STARTUP_MSG: version = 2.4.1
...
...
10/24/14 21:30:56 INFO common.Storage: Storage directory
/home/hadoop/hadoopinfra/hdfs/namenode has been successfully formatted.
10/24/14 21:30:56 INFO namenode.NNStorageRetentionManager: Going to
retain 1 images with txid >= 0
10/24/14 21:30:56 INFO util.ExitUtil: Exiting with status 0
10/24/14 21:30:56 INFO namenode.NameNode: SHUTDOWN_MSG:

/************************************************************
SHUTDOWN_MSG: Shutting down NameNode at localhost/192.168.1.11
************************************************************/

2 단계-Hadoop dfs 확인

다음 명령을 실행하여 Hadoop 파일 시스템을 시작합니다.

$ start-dfs.sh

예상 출력은 다음과 같습니다.

10/24/14 21:37:56
Starting namenodes on [localhost]
localhost: starting namenode, logging to /home/hadoop/hadoop-
2.4.1/logs/hadoop-hadoop-namenode-localhost.out
localhost: starting datanode, logging to /home/hadoop/hadoop-
2.4.1/logs/hadoop-hadoop-datanode-localhost.out
Starting secondary namenodes [0.0.0.0]

3 단계-Yarn 스크립트 확인

다음 명령은 yarn 스크립트를 시작하는 데 사용됩니다. 이 명령을 실행하면 yarn 데몬이 시작됩니다.

$ start-yarn.sh

예상 출력은 다음과 같습니다.

starting yarn daemons
starting resourcemanager, logging to /home/hadoop/hadoop-
2.4.1/logs/yarn-hadoop-resourcemanager-localhost.out
localhost: starting node manager, logging to /home/hadoop/hadoop-
2.4.1/logs/yarn-hadoop-nodemanager-localhost.out

4 단계-브라우저에서 Hadoop 액세스

Hadoop에 액세스하기위한 기본 포트 번호는 50070입니다. 다음 URL을 사용하여 브라우저에서 Hadoop 서비스를 가져옵니다.

http://localhost:50070/

다음 스크린 샷은 Hadoop 브라우저를 보여줍니다.

5 단계-클러스터의 모든 애플리케이션 확인

클러스터의 모든 애플리케이션에 액세스하기위한 기본 포트 번호는 8088입니다.이 서비스를 사용하려면 다음 URL을 사용하십시오.

http://localhost:8088/

다음 스크린 샷은 Hadoop 클러스터 브라우저를 보여줍니다.

이 장에서는 MapReduce 프로그래밍 작업에 관련된 클래스와 메서드를 자세히 살펴 보겠습니다. 우리는 주로 다음에 중점을 둘 것입니다.

  • JobContext 인터페이스
  • 직업 클래스
  • 매퍼 클래스
  • 감속기 클래스

JobContext 인터페이스

JobContext 인터페이스는 MapReduce에서 서로 다른 작업을 정의하는 모든 클래스에 대한 수퍼 인터페이스입니다. 작업이 실행되는 동안 작업에 제공되는 작업에 대한 읽기 전용보기를 제공합니다.

다음은 JobContext 인터페이스의 하위 인터페이스입니다.

S. 아니. 하위 인터페이스 설명
1. MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

매퍼에 제공되는 컨텍스트를 정의합니다.
2. ReduceContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

Reducer에 전달되는 컨텍스트를 정의합니다.

Job 클래스는 JobContext 인터페이스를 구현하는 기본 클래스입니다.

직업 클래스

Job 클래스는 MapReduce API에서 가장 중요한 클래스입니다. 이를 통해 사용자는 작업을 구성하고, 제출하고, 실행을 제어하고, 상태를 쿼리 할 수 ​​있습니다. set 메소드는 작업이 제출 될 때까지만 작동하며 이후에는 IllegalStateException이 발생합니다.

일반적으로 사용자는 응용 프로그램을 만들고 작업의 다양한 측면을 설명한 다음 작업을 제출하고 진행 상황을 모니터링합니다.

다음은 작업을 제출하는 방법의 예입니다.

// Create a new Job
Job job = new Job(new Configuration());
job.setJarByClass(MyJob.class);

// Specify various job-specific parameters
job.setJobName("myjob");
job.setInputPath(new Path("in"));
job.setOutputPath(new Path("out"));

job.setMapperClass(MyJob.MyMapper.class);
job.setReducerClass(MyJob.MyReducer.class);

// Submit the job, then poll for progress until the job is complete
job.waitForCompletion(true);

생성자

다음은 Job 클래스의 생성자 요약입니다.

S. 아니 생성자 요약
1 Job()
2 Job(구성 구성)
삼 Job(구성 구성, 문자열 작업 이름)

행동 양식

Job 클래스의 중요한 방법은 다음과 같습니다.

S. 아니 방법 설명
1 getJobName()

사용자 지정 작업 이름.
2 getJobState()

Job의 현재 상태를 반환합니다.
삼 isComplete()

작업이 완료되었는지 확인합니다.
4 setInputFormatClass()

작업에 대한 InputFormat을 설정합니다.
5 setJobName(String name)

사용자 지정 작업 이름을 설정합니다.
6 setOutputFormatClass()

작업의 출력 형식을 설정합니다.
7 setMapperClass(Class)

작업에 대한 매퍼를 설정합니다.
8 setReducerClass(Class)

작업에 대한 감속기를 설정합니다.
9 setPartitionerClass(Class)

작업에 대한 파티션 도구를 설정합니다.
10 setCombinerClass(Class)

작업에 대한 결합기를 설정합니다.

매퍼 클래스

Mapper 클래스는 Map 작업을 정의합니다. 입력 키-값 쌍을 중간 키-값 쌍 집합에 매핑합니다. 맵은 입력 레코드를 중간 레코드로 변환하는 개별 작업입니다. 변환 된 중간 레코드는 입력 레코드와 동일한 유형일 필요는 없습니다. 주어진 입력 쌍은 0 개 또는 많은 출력 쌍에 매핑 될 수 있습니다.

방법

mapMapper 클래스에서 가장 눈에 띄는 방법입니다. 구문은 아래에 정의되어 있습니다-

map(KEYIN key, VALUEIN value, org.apache.hadoop.mapreduce.Mapper.Context context)

이 메서드는 입력 분할의 각 키-값 쌍에 대해 한 번씩 호출됩니다.

감속기 클래스

Reducer 클래스는 MapReduce에서 Reduce 작업을 정의합니다. 키를 공유하는 중간 값 세트를 더 작은 값 세트로 줄입니다. Reducer 구현은 JobContext.getConfiguration () 메소드를 통해 작업의 구성에 액세스 할 수 있습니다. Reducer에는 Shuffle, Sort 및 Reduce의 세 가지 기본 단계가 있습니다.

  • Shuffle − Reducer는 네트워크를 통해 HTTP를 사용하여 각 매퍼에서 정렬 된 출력을 복사합니다.

  • Sort− 프레임 워크는 Reducer 입력을 키별로 병합 정렬합니다 (다른 Mapper가 동일한 키를 출력 할 수 있기 때문). 셔플 및 정렬 단계는 동시에 발생합니다. 즉, 출력을 가져 오는 동안 병합됩니다.

  • Reduce −이 단계에서는 정렬 된 입력의 각 <key, (값 모음)>에 대해 reduce (Object, Iterable, Context) 메서드가 호출됩니다.

방법

reduceReducer 클래스에서 가장 눈에 띄는 방법입니다. 구문은 아래에 정의되어 있습니다-

reduce(KEYIN key, Iterable<VALUEIN> values, org.apache.hadoop.mapreduce.Reducer.Context context)

이 메서드는 키-값 쌍 컬렉션의 각 키에 대해 한 번씩 호출됩니다.

MapReduce는 대규모 상용 하드웨어 클러스터에있는 대량의 데이터를 안정적인 방식으로 처리하기 위해 애플리케이션을 작성하는 데 사용되는 프레임 워크입니다. 이 장에서는 Java를 사용하여 Hadoop 프레임 워크에서 MapReduce를 작동하는 방법에 대해 설명합니다.

MapReduce 알고리즘

일반적으로 MapReduce 패러다임은 실제 데이터가있는 컴퓨터로 맵 축소 프로그램을 보내는 것을 기반으로합니다.

  • MapReduce 작업 중에 Hadoop은 Map 및 Reduce 작업을 클러스터의 적절한 서버로 보냅니다.

  • 프레임 워크는 작업 실행, 작업 완료 확인 및 노드 간 클러스터 주변의 데이터 복사와 같은 데이터 전달의 모든 세부 사항을 관리합니다.

  • 대부분의 컴퓨팅은 네트워크 트래픽을 줄이는 로컬 디스크의 데이터가있는 노드에서 이루어집니다.

  • 주어진 작업을 완료 한 후 클러스터는 적절한 결과를 형성하기 위해 데이터를 수집 및 축소하고 다시 Hadoop 서버로 보냅니다.

입력 및 출력 (Java Perspective)

MapReduce 프레임 워크는 키-값 쌍에서 작동합니다. 즉, 프레임 워크는 작업에 대한 입력을 키-값 쌍의 집합으로보고 다른 유형의 작업 출력으로 키-값 쌍 집합을 생성합니다.

키 및 값 클래스는 프레임 워크에서 직렬화 가능해야하므로 쓰기 가능 인터페이스를 구현해야합니다. 또한 키 클래스는 WritableComparable 인터페이스를 구현하여 프레임 워크 별 정렬을 용이하게해야합니다.

MapReduce 작업의 입력 및 출력 형식은 모두 키-값 쌍의 형식입니다.

(입력) <k1, v1>-> map-> <k2, v2>-> reduce-> <k3, v3> (출력).

입력 산출
지도 <k1, v1> 목록 (<k2, v2>)
줄이다 <k2, 목록 (v2)> 목록 (<k3, v3>)

MapReduce 구현

다음 표는 조직의 전력 소비에 관한 데이터를 보여줍니다. 이 표에는 5 년 연속 월간 전력 소비량과 연간 평균이 포함됩니다.

1 월 2 월 망치다 4 월 할 수있다 6 월 7 월 8 월 9 월 10 월 11 월 12 월 평균
1979 년 23 23 2 43 24 25 26 26 26 26 25 26 25
1980 년 26 27 28 28 28 30 31 31 31 30 30 30 29
1981 년 31 32 32 32 33 34 35 36 36 34 34 34 34
1984 년 39 38 39 39 39 41 42 43 40 39 38 38 40
1985 년 38 39 39 39 39 41 41 41 00 40 39 39 45

최대 사용 연도, 최소 사용 연도 등을 찾기 위해 주어진 테이블의 입력 데이터를 처리하는 응용 프로그램을 작성해야합니다. 이 작업은 필요한 출력을 생성하는 논리를 작성하고 작성된 응용 프로그램에 데이터를 전달하기 때문에 제한된 양의 레코드를 가진 프로그래머에게 쉽습니다.

이제 입력 데이터의 규모를 높여 보겠습니다. 특정주의 모든 대규모 산업의 전력 소비를 분석해야한다고 가정합니다. 이러한 대량 데이터를 처리하기 위해 애플리케이션을 작성할 때

  • 실행하는 데 많은 시간이 걸립니다.

  • 소스에서 네트워크 서버로 데이터를 이동할 때 네트워크 트래픽이 많이 발생합니다.

이러한 문제를 해결하기 위해 MapReduce 프레임 워크가 있습니다.

입력 데이터

위의 데이터는 다음과 같이 저장됩니다. sample.txt입력으로 제공됩니다. 입력 파일은 아래와 같습니다.

1979 년 23 23 2 43 24 25 26 26 26 26 25 26 25
1980 년 26 27 28 28 28 30 31 31 31 30 30 30 29
1981 년 31 32 32 32 33 34 35 36 36 34 34 34 34
1984 년 39 38 39 39 39 41 42 43 40 39 38 38 40
1985 년 38 39 39 39 39 41 41 41 00 40 39 39 45

예제 프로그램

샘플 데이터에 대한 다음 프로그램은 MapReduce 프레임 워크를 사용합니다.

package hadoop;

import java.util.*;
import java.io.IOException;
import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class ProcessUnits
{
   //Mapper class
   public static class E_EMapper extends MapReduceBase implements
   Mapper<LongWritable,  /*Input key Type */
   Text,                   /*Input value Type*/
   Text,                   /*Output key Type*/
   IntWritable>            /*Output value Type*/
   {
      //Map function
      public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException
      {
         String line = value.toString();
         String lasttoken = null;
         StringTokenizer s = new StringTokenizer(line,"\t");
         String year = s.nextToken();
         
         while(s.hasMoreTokens()){
            lasttoken=s.nextToken();
         }
         
         int avgprice = Integer.parseInt(lasttoken);
         output.collect(new Text(year), new IntWritable(avgprice));
      }
   }
   
   //Reducer class
	
   public static class E_EReduce extends MapReduceBase implements
   Reducer< Text, IntWritable, Text, IntWritable >
   {
      //Reduce function
      public void reduce(Text key, Iterator <IntWritable> values, OutputCollector>Text, IntWritable> output, Reporter reporter) throws IOException
      {
         int maxavg=30;
         int val=Integer.MIN_VALUE;
         while (values.hasNext())
         {
            if((val=values.next().get())>maxavg)
            {
               output.collect(key, new IntWritable(val));
            }
         }
      }
   }
	
   //Main function
	
   public static void main(String args[])throws Exception
   {
      JobConf conf = new JobConf(Eleunits.class);
		
      conf.setJobName("max_eletricityunits");
		
      conf.setOutputKeyClass(Text.class);
      conf.setOutputValueClass(IntWritable.class);
		
      conf.setMapperClass(E_EMapper.class);
      conf.setCombinerClass(E_EReduce.class);
      conf.setReducerClass(E_EReduce.class);
		
      conf.setInputFormat(TextInputFormat.class);
      conf.setOutputFormat(TextOutputFormat.class);
		
      FileInputFormat.setInputPaths(conf, new Path(args[0]));
      FileOutputFormat.setOutputPath(conf, new Path(args[1]));
		
      JobClient.runJob(conf);
   }
}

위의 프로그램을 ProcessUnits.java. 프로그램의 컴파일 및 실행은 다음과 같습니다.

ProcessUnits 프로그램의 컴파일 및 실행

Hadoop 사용자의 홈 디렉토리 (예 : / home / hadoop)에 있다고 가정 해 보겠습니다.

위의 프로그램을 컴파일하고 실행하려면 아래 단계를 따르십시오.

Step 1 − 컴파일 된 자바 클래스를 저장할 디렉토리를 생성하려면 다음 명령을 사용하십시오.

$ mkdir units

Step 2− MapReduce 프로그램을 컴파일하고 실행하는 데 사용되는 Hadoop-core-1.2.1.jar를 다운로드합니다. mvnrepository.com 에서 jar를 다운로드하십시오 . 다운로드 폴더가 / home / hadoop /이라고 가정하겠습니다.

Step 3 − 다음 명령을 사용하여 ProcessUnits.java 프로그램을 만들고 프로그램에 대한 jar를 만듭니다.

$ javac -classpath hadoop-core-1.2.1.jar -d units ProcessUnits.java
$ jar -cvf units.jar -C units/ .

Step 4 − 다음 명령은 HDFS에서 입력 디렉토리를 생성하는 데 사용됩니다.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 5 − 다음 명령은 이름이 지정된 입력 파일을 복사하는 데 사용됩니다. sample.txt HDFS의 입력 디렉토리에 있습니다.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/sample.txt input_dir

Step 6 − 다음 명령은 입력 디렉토리의 파일을 확인하는 데 사용됩니다.

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 7 − 다음 명령어는 입력 디렉토리에서 입력 파일을 가져와 Eleunit_max 응용 프로그램을 실행하는 데 사용됩니다.

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir

파일이 실행될 때까지 잠시 기다리십시오. 실행 후 출력에는 여러 입력 분할, Map 작업, Reducer 작업 등이 포함됩니다.

INFO mapreduce.Job: Job job_1414748220717_0002
completed successfully
14/10/31 06:02:52
INFO mapreduce.Job: Counters: 49

File System Counters
   
   FILE: Number of bytes read=61
   FILE: Number of bytes written=279400
   FILE: Number of read operations=0
   FILE: Number of large read operations=0
   FILE: Number of write operations=0

   HDFS: Number of bytes read=546
   HDFS: Number of bytes written=40
   HDFS: Number of read operations=9
   HDFS: Number of large read operations=0
   HDFS: Number of write operations=2 Job Counters
   
   Launched map tasks=2
   Launched reduce tasks=1
   Data-local map tasks=2
	
   Total time spent by all maps in occupied slots (ms)=146137
   Total time spent by all reduces in occupied slots (ms)=441
   Total time spent by all map tasks (ms)=14613
   Total time spent by all reduce tasks (ms)=44120
	
   Total vcore-seconds taken by all map tasks=146137
   Total vcore-seconds taken by all reduce tasks=44120
	
   Total megabyte-seconds taken by all map tasks=149644288
   Total megabyte-seconds taken by all reduce tasks=45178880

Map-Reduce Framework
   
   Map input records=5
	
   Map output records=5
   Map output bytes=45
   Map output materialized bytes=67
	
   Input split bytes=208
   Combine input records=5
   Combine output records=5
	
   Reduce input groups=5
   Reduce shuffle bytes=6
   Reduce input records=5
   Reduce output records=5
	
   Spilled Records=10
   Shuffled Maps =2
   Failed Shuffles=0
   Merged Map outputs=2
	
   GC time elapsed (ms)=948
   CPU time spent (ms)=5160
	
   Physical memory (bytes) snapshot=47749120
   Virtual memory (bytes) snapshot=2899349504
	
   Total committed heap usage (bytes)=277684224

File Output Format Counters

   Bytes Written=40

Step 8 − 다음 명령은 출력 폴더에서 결과 파일을 확인하는 데 사용됩니다.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

Step 9 − 다음 명령은 출력을 확인하는 데 사용됩니다. Part-00000파일. 이 파일은 HDFS에서 생성됩니다.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

다음은 MapReduce 프로그램에 의해 생성 된 출력입니다.

1981 년 34
1984 년 40
1985 년 45

Step 10 − 다음 명령은 HDFS에서 로컬 파일 시스템으로 출력 폴더를 복사하는 데 사용됩니다.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000/bin/hadoop dfs -get output_dir /home/hadoop

파티 셔 너는 입력 데이터 세트를 처리 할 때 조건처럼 작동합니다. 파티션 단계는 매핑 단계 이후와 축소 단계 이전에 발생합니다.

파티 셔 너의 수는 리듀서의 수와 같습니다. 즉, 파티 셔 너는 리듀서의 수에 따라 데이터를 나눕니다. 따라서 단일 파티 셔너에서 전달 된 데이터는 단일 Reducer에 의해 처리됩니다.

파티 셔너

파티 셔 너는 중간 맵 출력의 키-값 쌍을 분할합니다. 해시 함수처럼 작동하는 사용자 정의 조건을 사용하여 데이터를 분할합니다. 총 파티션 수는 작업에 대한 Reducer 작업 수와 동일합니다. 파티 셔 너가 어떻게 작동하는지 이해하기위한 예를 들어 보겠습니다.

MapReduce 파티션 도구 구현

편의를 위해 다음 데이터가 포함 된 Employee라는 작은 테이블이 있다고 가정합니다. 이 샘플 데이터를 입력 데이터 세트로 사용하여 파티 셔 너가 작동하는 방식을 보여줍니다.

신분증 이름 나이 성별 봉급
1201 고팔 45 남성 50,000
1202 Manisha 40 여자 50,000
1203 년 칼릴 34 남성 30,000
1204 프라 산스 30 남성 30,000
1205 년 키란 20 남성 40,000
1206 년 Laxmi 25 여자 35,000
1207 바비 아 20 여자 15,000
1208 년 레 쉬마 19 여자 15,000
1209 년 Kranthi 22 남성 22,000
1210 년 Satish 24 남성 25,000
1211 크리슈나 25 남성 25,000
1212 년 아르 샤드 28 남성 20,000
1213 년 Lavanya 18 여자 8,000

다른 연령대 (예 : 20 세 미만, 21 세 ~ 30 세, 30 세 이상)에서 성별별로 가장 높은 급여를받는 직원을 찾기 위해 입력 데이터 세트를 처리하는 애플리케이션을 작성해야합니다.

입력 데이터

위의 데이터는 다음과 같이 저장됩니다. input.txt "/ home / hadoop / hadoopPartitioner"디렉토리에서 입력으로 제공됩니다.

1201 고팔 45 남성 50000
1202 Manisha 40 여자 51000
1203 년 Khaleel 34 남성 30000
1204 프라 산스 30 남성 31000
1205 년 키란 20 남성 40000
1206 년 Laxmi 25 여자 35000
1207 바비 아 20 여자 15000
1208 년 레 쉬마 19 여자 14000
1209 년 Kranthi 22 남성 22000
1210 년 Satish 24 남성 25000
1211 크리슈나 25 남성 26000
1212 년 아르 샤드 28 남성 20000
1213 년 Lavanya 18 여자 8000

주어진 입력에 따라 프로그램에 대한 알고리즘 설명은 다음과 같습니다.

작업 매핑

맵 태스크는 텍스트 파일에 텍스트 데이터가있는 동안 키-값 쌍을 입력으로받습니다. 이지도 작업에 대한 입력은 다음과 같습니다.

Input − 키는 "모든 특수 키 + 파일 이름 + 줄 번호"(예 : 키 = @ input1)와 같은 패턴이고 값은 해당 줄의 데이터입니다 (예 : 값 = 1201 \ t gopal \ t 45 \ t 남성 \ t 50000).

Method −이지도 작업의 동작은 다음과 같습니다 −

  • 읽기 value (레코드 데이터), 문자열의 인수 목록에서 입력 값으로 제공됩니다.

  • split 함수를 사용하여 성별을 분리하고 문자열 변수에 저장합니다.

String[] str = value.toString().split("\t", -3);
String gender=str[3];

  • 성별 정보 및 기록 데이터 보내기 value 매핑 작업에서 출력 키-값 쌍으로 partition task.

context.write(new Text(gender), new Text(value));

  • 텍스트 파일의 모든 레코드에 대해 위의 모든 단계를 반복합니다.

Output − 성별 데이터와 레코드 데이터 값을 키-값 쌍으로 가져옵니다.

파티 셔너 작업

파티 셔너 태스크는 맵 태스크의 키-값 쌍을 입력으로받습니다. 파티션은 데이터를 세그먼트로 나누는 것을 의미합니다. 주어진 파티션 조건부 기준에 따라 입력 키-값 쌍 데이터는 연령 기준에 따라 세 부분으로 나눌 수 있습니다.

Input − 키-값 쌍 모음의 전체 데이터.

키 = 레코드의 성별 필드 값.

value = 해당 성별의 전체 레코드 데이터 값.

Method − 파티션 로직의 과정은 다음과 같습니다.

  • 입력 키-값 쌍에서 연령 필드 값을 읽습니다.
String[] str = value.toString().split("\t");
int age = Integer.parseInt(str[2]);

  • 다음 조건으로 연령 값을 확인하십시오.

    • 20 세 이하
    • 나이 20 세 이상 30 세 이하.
    • 30 세 이상.
if(age<=20)
{
   return 0;
}
else if(age>20 && age<=30)
{
   return 1 % numReduceTasks;
}
else
{
   return 2 % numReduceTasks;
}

Output− 키-값 쌍의 전체 데이터는 키-값 쌍의 세 가지 모음으로 나뉩니다. Reducer는 각 컬렉션에서 개별적으로 작동합니다.

작업 감소

파티 셔너 작업의 수는 리듀서 작업의 수와 같습니다. 여기에는 세 개의 파티 셔너 작업이 있으므로 실행할 세 개의 Reducer 작업이 있습니다.

Input − Reducer는 서로 다른 키-값 쌍 모음으로 세 번 실행됩니다.

키 = 레코드의 성별 필드 값.

값 = 해당 성별의 전체 기록 데이터.

Method − 각 컬렉션에는 다음과 같은 로직이 적용됩니다.

  • 각 레코드의 급여 필드 값을 읽습니다.
String [] str = val.toString().split("\t", -3);
Note: str[4] have the salary field value.

  • max 변수로 급여를 확인하십시오. str [4]가 최대 급여이면 str [4]를 max에 할당하고 그렇지 않으면 단계를 건너 뜁니다.

if(Integer.parseInt(str[4])>max)
{
   max=Integer.parseInt(str[4]);
}

  • 각 키 컬렉션에 대해 1 단계와 2 단계를 반복합니다 (남성 및 여성이 키 컬렉션 임). 이 세 단계를 실행하면 남성 키 컬렉션에서 최대 급여 하나와 여성 키 컬렉션에서 최대 급여 하나를 찾을 수 있습니다.

context.write(new Text(key), new IntWritable(max));

Output− 마지막으로, 서로 다른 연령 그룹의 세 가지 컬렉션에서 키-값 쌍 데이터 세트를 얻습니다. 여기에는 각 연령 그룹의 남성 컬렉션의 최대 급여와 여성 컬렉션의 최대 급여가 포함됩니다.

Map, Partitioner 및 Reduce 작업을 실행 한 후 세 가지 키-값 쌍 데이터 컬렉션이 출력으로 세 개의 다른 파일에 저장됩니다.

세 가지 작업은 모두 MapReduce 작업으로 처리됩니다. 이러한 작업의 다음 요구 사항 및 사양은 구성에 지정되어야합니다-

  • 직업 이름
  • 키와 값의 입력 및 출력 형식
  • Map, Reduce 및 Partitioner 작업을위한 개별 클래스
Configuration conf = getConf();

//Create Job
Job job = new Job(conf, "topsal");
job.setJarByClass(PartitionerExample.class);

// File Input and Output paths
FileInputFormat.setInputPaths(job, new Path(arg[0]));
FileOutputFormat.setOutputPath(job,new Path(arg[1]));

//Set Mapper class and Output format for key-value pair.
job.setMapperClass(MapClass.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);

//set partitioner statement
job.setPartitionerClass(CaderPartitioner.class);

//Set Reducer class and Input/Output format for key-value pair.
job.setReducerClass(ReduceClass.class);

//Number of Reducer tasks.
job.setNumReduceTasks(3);

//Input and Output format for data
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

예제 프로그램

다음 프로그램은 MapReduce 프로그램에서 주어진 기준에 대한 파티 셔 너를 구현하는 방법을 보여줍니다.

package partitionerexample;

import java.io.*;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;

import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;

import org.apache.hadoop.util.*;

public class PartitionerExample extends Configured implements Tool
{
   //Map class
	
   public static class MapClass extends Mapper<LongWritable,Text,Text,Text>
   {
      public void map(LongWritable key, Text value, Context context)
      {
         try{
            String[] str = value.toString().split("\t", -3);
            String gender=str[3];
            context.write(new Text(gender), new Text(value));
         }
         catch(Exception e)
         {
            System.out.println(e.getMessage());
         }
      }
   }
   
   //Reducer class
	
   public static class ReduceClass extends Reducer<Text,Text,Text,IntWritable>
   {
      public int max = -1;
      public void reduce(Text key, Iterable <Text> values, Context context) throws IOException, InterruptedException
      {
         max = -1;
			
         for (Text val : values)
         {
            String [] str = val.toString().split("\t", -3);
            if(Integer.parseInt(str[4])>max)
            max=Integer.parseInt(str[4]);
         }
			
         context.write(new Text(key), new IntWritable(max));
      }
   }
   
   //Partitioner class
	
   public static class CaderPartitioner extends
   Partitioner < Text, Text >
   {
      @Override
      public int getPartition(Text key, Text value, int numReduceTasks)
      {
         String[] str = value.toString().split("\t");
         int age = Integer.parseInt(str[2]);
         
         if(numReduceTasks == 0)
         {
            return 0;
         }
         
         if(age<=20)
         {
            return 0;
         }
         else if(age>20 && age<=30)
         {
            return 1 % numReduceTasks;
         }
         else
         {
            return 2 % numReduceTasks;
         }
      }
   }
   
   @Override
   public int run(String[] arg) throws Exception
   {
      Configuration conf = getConf();
		
      Job job = new Job(conf, "topsal");
      job.setJarByClass(PartitionerExample.class);
		
      FileInputFormat.setInputPaths(job, new Path(arg[0]));
      FileOutputFormat.setOutputPath(job,new Path(arg[1]));
		
      job.setMapperClass(MapClass.class);
		
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(Text.class);
      
      //set partitioner statement
		
      job.setPartitionerClass(CaderPartitioner.class);
      job.setReducerClass(ReduceClass.class);
      job.setNumReduceTasks(3);
      job.setInputFormatClass(TextInputFormat.class);
		
      job.setOutputFormatClass(TextOutputFormat.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);
		
      System.exit(job.waitForCompletion(true)? 0 : 1);
      return 0;
   }
   
   public static void main(String ar[]) throws Exception
   {
      int res = ToolRunner.run(new Configuration(), new PartitionerExample(),ar);
      System.exit(0);
   }
}

위 코드를 다른 이름으로 저장 PartitionerExample.java"/ home / hadoop / hadoopPartitioner"에 있습니다. 프로그램의 컴파일 및 실행은 다음과 같습니다.

컴파일 및 실행

Hadoop 사용자의 홈 디렉토리 (예 : / home / hadoop)에 있다고 가정 해 보겠습니다.

위의 프로그램을 컴파일하고 실행하려면 아래 단계를 따르십시오.

Step 1− MapReduce 프로그램을 컴파일하고 실행하는 데 사용되는 Hadoop-core-1.2.1.jar를 다운로드합니다. mvnrepository.com 에서 jar를 다운로드 할 수 있습니다 .

다운로드 한 폴더가“/ home / hadoop / hadoopPartitioner”라고 가정하겠습니다.

Step 2 − 다음 명령은 프로그램 컴파일에 사용됩니다. PartitionerExample.java 프로그램을위한 jar를 생성합니다.

$ javac -classpath hadoop-core-1.2.1.jar -d ProcessUnits.java $ jar -cvf PartitionerExample.jar -C .

Step 3 − 다음 명령을 사용하여 HDFS에서 입력 디렉토리를 생성합니다.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 4 − 다음 명령을 사용하여 이름이 지정된 입력 파일을 복사합니다. input.txt HDFS의 입력 디렉토리에 있습니다.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/hadoopPartitioner/input.txt input_dir

Step 5 − 다음 명령을 사용하여 입력 디렉토리의 파일을 확인합니다.

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 6 − 다음 명령을 사용하여 입력 디렉토리에서 입력 파일을 가져와 최고 급여 응용 프로그램을 실행합니다.

$HADOOP_HOME/bin/hadoop jar PartitionerExample.jar partitionerexample.PartitionerExample input_dir/input.txt output_dir

파일이 실행될 때까지 잠시 기다리십시오. 실행 후 출력에는 여러 입력 분할, 매핑 작업 및 Reducer 작업이 포함됩니다.

15/02/04 15:19:51 INFO mapreduce.Job: Job job_1423027269044_0021 completed successfully
15/02/04 15:19:52 INFO mapreduce.Job: Counters: 49

File System Counters

   FILE: Number of bytes read=467
   FILE: Number of bytes written=426777
   FILE: Number of read operations=0
   FILE: Number of large read operations=0
   FILE: Number of write operations=0
	
   HDFS: Number of bytes read=480
   HDFS: Number of bytes written=72
   HDFS: Number of read operations=12
   HDFS: Number of large read operations=0
   HDFS: Number of write operations=6
	
Job Counters

   Launched map tasks=1
   Launched reduce tasks=3
	
   Data-local map tasks=1
	
   Total time spent by all maps in occupied slots (ms)=8212
   Total time spent by all reduces in occupied slots (ms)=59858
   Total time spent by all map tasks (ms)=8212
   Total time spent by all reduce tasks (ms)=59858
	
   Total vcore-seconds taken by all map tasks=8212
   Total vcore-seconds taken by all reduce tasks=59858
	
   Total megabyte-seconds taken by all map tasks=8409088
   Total megabyte-seconds taken by all reduce tasks=61294592
	
Map-Reduce Framework

   Map input records=13
   Map output records=13
   Map output bytes=423
   Map output materialized bytes=467
	
   Input split bytes=119
	
   Combine input records=0
   Combine output records=0
	
   Reduce input groups=6
   Reduce shuffle bytes=467
   Reduce input records=13
   Reduce output records=6
	
   Spilled Records=26
   Shuffled Maps =3
   Failed Shuffles=0
   Merged Map outputs=3
   GC time elapsed (ms)=224
   CPU time spent (ms)=3690
	
   Physical memory (bytes) snapshot=553816064
   Virtual memory (bytes) snapshot=3441266688
	
   Total committed heap usage (bytes)=334102528
	
Shuffle Errors

   BAD_ID=0
   CONNECTION=0
   IO_ERROR=0
	
   WRONG_LENGTH=0
   WRONG_MAP=0
   WRONG_REDUCE=0
	
File Input Format Counters

   Bytes Read=361
	
File Output Format Counters

   Bytes Written=72

Step 7 − 다음 명령을 사용하여 출력 폴더에서 결과 파일을 확인합니다.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

프로그램에서 3 개의 파티 셔 너와 3 개의 Reducer를 사용하고 있기 때문에 3 개의 파일에서 출력을 찾을 수 있습니다.

Step 8 − 다음 명령을 사용하여 출력을 확인하십시오. Part-00000파일. 이 파일은 HDFS에서 생성됩니다.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

Output in Part-00000

Female   15000
Male     40000

다음 명령을 사용하여 출력을 확인하십시오. Part-00001 파일.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00001

Output in Part-00001

Female   35000
Male    31000

다음 명령을 사용하여 출력을 확인하십시오. Part-00002 파일.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00002

Output in Part-00002

Female  51000
Male   50000

결합기, 일명 semi-reducer, Map 클래스의 입력을 수락 한 후 출력 키-값 쌍을 Reducer 클래스에 전달하여 작동하는 선택적 클래스입니다.

Combiner의 주요 기능은 동일한 키로 맵 출력 레코드를 요약하는 것입니다. 결합기의 출력 (키-값 수집)은 네트워크를 통해 실제 Reducer 작업에 입력으로 전송됩니다.

결합기

Combiner 클래스는 Map 클래스와 Reduce 클래스 사이에서 사용되어 Map과 Reduce 사이의 데이터 전송량을 줄입니다. 일반적으로 맵 작업의 출력이 크고 축소 작업으로 전송되는 데이터가 높습니다.

다음 MapReduce 작업 다이어그램은 COMBINER PHASE를 보여줍니다.

컴 바이 너는 어떻게 작동합니까?

다음은 MapReduce Combiner의 작동 방식에 대한 간략한 요약입니다.

  • 컴 바이 너에는 미리 정의 된 인터페이스가 없으며 Reducer 인터페이스의 reduce () 메서드를 구현해야합니다.

  • 컴 바이 너는 각 맵 출력 키에서 작동합니다. Reducer 클래스와 동일한 출력 키-값 유형이 있어야합니다.

  • 결합기는 원래 Map 출력을 대체하기 때문에 대규모 데이터 세트에서 요약 정보를 생성 할 수 있습니다.

Combiner는 선택 사항이지만 Reduce 단계를 위해 데이터를 여러 그룹으로 분리하는 데 도움이되므로 처리가 더 쉽습니다.

MapReduce Combiner 구현

다음 예제는 결합기에 대한 이론적 아이디어를 제공합니다. 다음과 같은 입력 텍스트 파일이 있다고 가정하겠습니다.input.txt MapReduce 용.

What do you mean by Object
What do you know about Java
What is Java Virtual Machine
How Java enabled High Performance

Combiner를 사용한 MapReduce 프로그램의 중요한 단계는 아래에서 설명합니다.

레코드 리더

이것은 레코드 리더가 입력 텍스트 파일의 모든 행을 텍스트로 읽고 키-값 쌍으로 출력을 산출하는 MapReduce의 첫 번째 단계입니다.

Input − 입력 파일에서 한 줄씩 텍스트.

Output− 키-값 쌍을 형성합니다. 다음은 예상되는 키-값 쌍의 집합입니다.

<1, What do you mean by Object>
<2, What do you know about Java>
<3, What is Java Virtual Machine>
<4, How Java enabled High Performance>

지도 단계

Map 단계는 레코드 리더에서 입력을 받아 처리하고 출력을 다른 키-값 쌍 세트로 생성합니다.

Input − 다음 키-값 쌍은 레코드 리더에서 가져온 입력입니다.

<1, What do you mean by Object>
<2, What do you know about Java>
<3, What is Java Virtual Machine>
<4, How Java enabled High Performance>

Map 단계는 각 키-값 쌍을 읽고 StringTokenizer를 사용하여 값에서 각 단어를 나누고 각 단어를 키로 취급하고 해당 단어의 개수를 값으로 처리합니다. 다음 코드 스 니펫은 Mapper 클래스와지도 함수를 보여줍니다.

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

Output − 예상 출력은 다음과 같습니다 −

<What,1> <do,1> <you,1> <mean,1> <by,1> <Object,1>
<What,1> <do,1> <you,1> <know,1> <about,1> <Java,1>
<What,1> <is,1> <Java,1> <Virtual,1> <Machine,1>
<How,1> <Java,1> <enabled,1> <High,1> <Performance,1>

결합기 단계

Combiner 단계는 Map 단계에서 각 키-값 쌍을 가져 와서 처리하고 다음과 같이 출력을 생성합니다. key-value collection 한 쌍.

Input − 다음 키-값 쌍은 매핑 단계에서 가져온 입력입니다.

<What,1> <do,1> <you,1> <mean,1> <by,1> <Object,1>
<What,1> <do,1> <you,1> <know,1> <about,1> <Java,1>
<What,1> <is,1> <Java,1> <Virtual,1> <Machine,1>
<How,1> <Java,1> <enabled,1> <High,1> <Performance,1>

Combiner 단계는 각 키-값 쌍을 읽고 공통 단어를 키로, 값을 컬렉션으로 결합합니다. 일반적으로 Combiner의 코드 및 작업은 Reducer의 코드와 작업과 유사합니다. 다음은 Mapper, Combiner 및 Reducer 클래스 선언에 대한 코드 조각입니다.

job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);

Output − 예상 출력은 다음과 같습니다 −

<What,1,1,1> <do,1,1> <you,1,1> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,1,1,1>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

감속기 단계

Reducer 단계는 Combiner 단계에서 각 키-값 컬렉션 쌍을 가져 와서 처리하고 출력을 키-값 쌍으로 전달합니다. Combiner 기능은 Reducer와 동일합니다.

Input − 다음 키-값 쌍은 결합기 단계에서 가져온 입력입니다.

<What,1,1,1> <do,1,1> <you,1,1> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,1,1,1>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

Reducer 단계는 각 키-값 쌍을 읽습니다. 다음은 Combiner의 코드 스 니펫입니다.

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

Output − Reducer 단계에서 예상되는 출력은 다음과 같습니다. −

<What,3> <do,2> <you,2> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,3>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

기록 작가

이것은 레코드 작성자가 Reducer 단계에서 모든 키-값 쌍을 작성하고 출력을 텍스트로 보내는 MapReduce의 마지막 단계입니다.

Input − 출력 형식과 함께 Reducer 단계의 각 키-값 쌍.

Output− 텍스트 형식으로 키-값 쌍을 제공합니다. 다음은 예상 출력입니다.

What           3
do             2
you            2
mean           1
by             1
Object         1
know           1
about          1
Java           3
is             1
Virtual        1
Machine        1
How            1
enabled        1
High           1
Performance    1

예제 프로그램

다음 코드 블록은 프로그램의 단어 수를 계산합니다.

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

위 프로그램을 다른 이름으로 저장 WordCount.java. 프로그램의 컴파일 및 실행은 다음과 같습니다.

컴파일 및 실행

Hadoop 사용자의 홈 디렉토리 (예 : / home / hadoop)에 있다고 가정 해 보겠습니다.

위의 프로그램을 컴파일하고 실행하려면 아래 단계를 따르십시오.

Step 1 − 컴파일 된 자바 클래스를 저장할 디렉토리를 생성하려면 다음 명령을 사용하십시오.

$ mkdir units

Step 2− MapReduce 프로그램을 컴파일하고 실행하는 데 사용되는 Hadoop-core-1.2.1.jar를 다운로드합니다. mvnrepository.com 에서 jar를 다운로드 할 수 있습니다 .

다운로드 한 폴더가 / home / hadoop /이라고 가정하겠습니다.

Step 3 − 다음 명령을 사용하여 WordCount.java 프로그램을 만들고 프로그램에 대한 jar를 만듭니다.

$ javac -classpath hadoop-core-1.2.1.jar -d units WordCount.java
$ jar -cvf units.jar -C units/ .

Step 4 − 다음 명령을 사용하여 HDFS에서 입력 디렉토리를 생성합니다.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 5 − 다음 명령을 사용하여 이름이 지정된 입력 파일을 복사합니다. input.txt HDFS의 입력 디렉토리에 있습니다.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/input.txt input_dir

Step 6 − 다음 명령을 사용하여 입력 디렉토리의 파일을 확인합니다.

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 7 − 다음 명령을 사용하여 입력 디렉토리에서 입력 파일을 가져 와서 단어 수 응용 프로그램을 실행합니다.

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir

파일이 실행될 때까지 잠시 기다리십시오. 실행 후 출력에는 여러 입력 분할, Map 작업 및 Reducer 작업이 포함됩니다.

Step 8 − 다음 명령을 사용하여 출력 폴더에서 결과 파일을 확인합니다.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

Step 9 − 다음 명령을 사용하여 출력을 확인하십시오. Part-00000파일. 이 파일은 HDFS에서 생성됩니다.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

다음은 MapReduce 프로그램에서 생성 된 출력입니다.

What           3
do             2
you            2
mean           1
by             1
Object         1
know           1
about          1
Java           3
is             1
Virtual        1
Machine        1
How            1
enabled        1
High           1
Performance    1

이 장에서는 HDFS 및 MapReduce 관리를 모두 포함하는 Hadoop 관리에 대해 설명합니다.

  • HDFS 관리에는 HDFS 파일 구조, 위치 및 업데이트 된 파일 모니터링이 포함됩니다.

  • MapReduce 관리에는 애플리케이션 목록 모니터링, 노드 구성, 애플리케이션 상태 등이 포함됩니다.

HDFS 모니터링

HDFS (Hadoop 분산 파일 시스템)에는 사용자 디렉터리, 입력 파일 및 출력 파일이 포함됩니다. MapReduce 명령을 사용하십시오.put 과 get, 저장 및 검색을 위해.

"/ $ HADOOP_HOME / sbin"에 "start-all.sh"명령을 전달하여 Hadoop 프레임 워크 (데몬)를 시작한 후 다음 URL을 "http : // localhost : 50070"브라우저에 전달합니다. 브라우저에 다음 화면이 표시되어야합니다.

다음 스크린 샷은 HDFS 탐색 방법을 보여줍니다.

다음 스크린 샷은 HDFS의 파일 구조를 보여줍니다. "/ user / hadoop"디렉토리에있는 파일을 보여줍니다.

다음 스크린 샷은 클러스터의 데이터 노드 정보를 보여줍니다. 여기에서 구성 및 용량이있는 하나의 노드를 찾을 수 있습니다.

MapReduce 작업 모니터링

MapReduce 애플리케이션은 작업 모음입니다 (Map 작업, Combiner, Partitioner 및 Reduce 작업). 다음을 모니터링하고 유지해야합니다.

  • 애플리케이션이 적합한 데이터 노드 구성.
  • 애플리케이션 당 사용 된 데이터 노드 및 리소스 수입니다.

이 모든 것을 모니터링하려면 사용자 인터페이스가 있어야합니다. “/ $ HADOOP_HOME / sbin”에“start-all.sh”명령을 전달하여 Hadoop 프레임 워크를 시작한 후 브라우저“http : // localhost : 8080”에 다음 URL을 전달합니다. 브라우저에 다음 화면이 표시되어야합니다.

위의 스크린 샷에서 손 모양 포인터는 애플리케이션 ID에 있습니다. 브라우저에서 다음 화면을 찾으려면 클릭하십시오. 다음을 설명합니다-

  • 현재 애플리케이션이 실행중인 사용자

  • 응용 프로그램 이름

  • 해당 응용 프로그램의 유형

  • 현재 상태, 최종 상태

  • 신청 시작 시간, 경과 (완료 시간), 모니터링 시점에 완료된 경우

  • 이 애플리케이션의 기록, 즉 로그 정보

  • 마지막으로 노드 정보, 즉 애플리케이션 실행에 참여한 노드입니다.

다음 스크린 샷은 특정 응용 프로그램의 세부 정보를 보여줍니다.

다음 스크린 샷은 현재 실행중인 노드 정보를 설명합니다. 여기서 스크린 샷에는 노드가 하나만 포함되어 있습니다. 핸드 포인터는 실행중인 노드의 로컬 호스트 주소를 표시합니다.

ja/tutorial
es/tutorial
de/tutorial
fr/tutorial
th/tutorial
pt/tutorial
ru/tutorial
vi/tutorial
it/tutorial
ko/tutorial
tr/tutorial
id/tutorial
pl/tutorial
hi/tutorial
japost
espost
depost
frpost
thpost
ptpost
rupost
vipost
itpost
kopost
trpost
idpost
plpost
hipost

© Copyright 2021 - 2024 | All Rights Reserved