Hadoop_2

hadoop_img_4

MapReduce가 하둡 클러스터에서 동작하는 방식

hadoop_img_5

Job Tracker

- 리소스 관리 기능 : 실제로 분석이 이루어지는 노드들(Task Tracker)의 리소스(가용 스토리지 등)가 이용 가능하지, 얼마나 사용되고 있는지 등을 관리

- 실행 관리 기능 : 실제로 분석을 실행하는 MapReduce Job을 배포하고 스케쥴링하고 모니터링하는 기능

Task Tracker

- Job Tracker에서 일을 받아와서 실행하고 실행 현황을 다시 보고한다.

★ MapReduce 프레임워크

MapReduce란: 2004년 구글랩에서 발표한 ‘MapReduce : Simplified Data Processing on Large Cluster’란 논문을 바탕으로 작성된 마스터/슬레이브 구조의 분산처리 시스템.

MapReduce 의 특징

- 데이터가 있는 서버로 코드를 전송: 소스코드보다 데이터의 크기가 훨씬 크기 때문에 처리할 데이터가 있는 서버로 코드를 전송함으로써 실행속도를 빠르게 한다.
- 데이터를 키/밸류 데이터셋의 변환으로 처리: 기본적으로 데이터를 키/밸류 페어로 처리하는데 크게 두가지 스텝으로 데이터를 처리한다.

가. Map Task

입력데이터를 여러 개의 조각으로 나눈 후 그것들을 데이터 조각 수 만큼 각 서버에서 병렬처리하게 된다. 이를 Map Task라고 한다.

Map Task의 입력, 출력 모두 키/밸류 페어가 된다.

나. Reduce Task: Map Task에서 각 서버에서 병렬처리한 결과를 모아서 최종 처리하는 과정이다. Reduce Task 역시 하나 이상의 서버에서 실행 가능하다. 이 스텝에서는 맵 테스크들에서 나온 레코드들 중에서 같은 키를 갖는 페어들을 묶어서 하나의 레코드를 만든다. 만들어진 레코드를 리듀스 태스크로 넘기게 되고 리듀스 태스크는 그 레코드를 처리하여 또 다른 키/밸류 페어를 만들어서 프로그래머가 지정한 HDFS상의 위치에 저장한다.

MapReduce 프레임워크가 같은 키를 갖는 페어들을 묶는 것을 알아서 해주기 때문에 프로그래머는 신경쓸 필요가 없고 묶인 키/밸류 페어들의 집합을 하고자 하는 일에 맞춰 처리하기만 하면 된다.

- Share Nothing: MapReduce는 병렬성이 아주 높다. 그 이유는 맵 태스크끼리 혹은 리듀스 태스크끼리 서로 의존성이 없기 때문이다.
- 오프라인 배치 처리에 적합: MapReduce 특징에 따라 대용량 데이터의 오프라인 배치처리를 위한 시스템이지 리얼타임 처리를 위한 시스템은 아니다.
MapReduce 흐름: 어느 MapReduce 프로그램이건 맵과 리듀스를 구현하여야 한다.

기본적으로 MapReduce 프로그램은 다음과 같은 특성을 지닌다.

- 맵과 리듀스의 두 단계로 구성

- 맵과 리듀스 모두 입력, 출력 데이터가 키와 밸류로 구성된다.

Map 단계에서는 주어진 키와 밸류를 새로운 키와 밸류로 변환한다

모든 입력 레코드들이 맵을 통해 처리되었으면 리듀스 작업이 시작된다.

맵에서 출력된 레코드들에서 같은 키 값을 갖는 레코드들을 모아서 리듀스에서 하나의 입력으로 들어간다.

hadoop_img_6

모두 세 개의 입력 텍스트가 있어서 세 개의 맵 태스크가 존재하고 리듀스 태스크는 2개라고 가정해보자.

맵에서는 들어오는 텍스트를 단어로 나눠서 각 단어가 들어오는 대로 빈도수를 1로해서 Key, Value쌍으로 출력한다.

리듀스에서는 모든 맵 태스크들의 출력들을 모아서 단어를 키로 하고 단어의 빈도수 리스트를 밸류로 만들어서 리듀스의 입력으로 지정한다.

그러면 리듀스는 간단하게 리듀스의 밸류로 들어온 빈도수 리스트를 스캔해서 빈도수를 다 더한 값을 구한다

리듀스의 출력은 결국 해당 단어가 키가되고 밸류는 그 단어의 총 빈도수가 된다.

hadoop_img_7

[실습]

Project : hadoop_mapreduce

Package : WordCount

Class : WordCountMapper.java

WordCountReducer.java

WordCount.java

1.input.txt 파일 작성

[hadoop@master jar]$ vi input.txt
[hadoop@master jar]$ cat input.txt

2.파일 올리기

[hadoop@master jar]$ hadoop fs -put input.txt input.txt
[hadoop@master jar]$ hadoop fs -ls

3.소스 작성

4.JAR 생성 - WordCount.jar

5.WordCount 빌드

입력파일 : input.txt

출력폴더 : output

[hadoop@master jar]$ hadoop jar WordCount.jar WordCount.WordCount input.txt output

6.결과

[hadoop@master jar]$ hadoop fs -ls
[hadoop@master jar]$ hadoop fs -ls output
Found 2 items
-rw-r--r--  1 hadoop supergroup     0 2017-02-22 17:35 output/_SUCCESS
-rw-r--r--  1 hadoop supergroup     58 2017-02-22 17:35 output/part-r-00000
[hadoop@master jar]$ 
[hadoop@master jar]$ hadoop fs -cat output/part-r-00000

[실습]

Project : hadoop_mapreduce

Package : CharCount

Class : CharCountMapper.java

CharCountReducer.java

CharCount.java

입력파일 : input_str.txt

출력폴더 : output_str

[hadoop@master jar]$ vi input_str.txt
ILOVEYOUILIKEYOU

Stiring 클래스의 charAt()를 이용하세요

I 1

L 1

O 1

V 1

E 1

Y 1

O 1

U 1

[실습]

Project : hadoop_mapreduce

Package : NewsCount

Class : NewsCountMapper.java

NewsCountReducer.java

NewsCount.java

입력파일 : input_news.txt

출력폴더 : output_news

[hadoop@master jar]$ vi input_news.txt
인터넷의 뉴스(영문)

A~Z까지(대문자/소문자 가리지 말 것) 각각 몇 개가 있는지 출력하시오

A 1

B 3

C 10

…

Z 7

★ 항공 출발 지연 데이터 분석

1.실습 데이터 내려받기

http://stat-computing.org/dataexpo/2009 접속 - Download the data 클릭

1987, 1988, 1989 년도 데이터를 다운로드 받는다.

2.다운로드 받은 파일을 workspace로 복사

3.파일 압축 해제

[hadoop@master workspace]$ bzip2 –d 1987.csv.bz2
[hadoop@master workspace]$ bzip2 –d 1988.csv.bz2
[hadoop@master workspace]$ bzip2 –d 1989.csv.bz2

또는 

[hadoop@master workspace]$ bzip2 -d *.bz2

4.첫째줄은 제목라인이다. 삭제하자.

[hadoop@master workspace]$ more 1987.csv
[hadoop@master workspace]$ sed –e ‘1d’ 1987.csv > 1987_new.csv
( 1988, 1989 마찬가지 )

5.HDFS에 분석용 파일 업로드하기.

[hadoop@master workspace]$ hadoop fs -ls
[hadoop@master workspace]$ hadoop fs –mkdir airline_input
[hadoop@master workspace]$ hadoop fs –put *_new.csv airline_input
[hadoop@master workspace]$ hadoop fs -ls
[hadoop@master workspace]$ hadoop fs -ls airline_input
[hadoop@master workspace]$ hadoop fs -ls
[hadoop@master workspace]$ hadoop fs –mkdir airline_input
[hadoop@master workspace]$ hadoop fs –put *_new.csv airline_input
[hadoop@master workspace]$ hadoop fs -ls
[hadoop@master workspace]$ hadoop fs -ls airline_input

6.AirlinePerformanceParser.java 구현

통계 데이터는 콤마(,)단위로 데이터가 저장되어 있다.

한 줄의 통계 데이터에 해당하는 객체를 생성하자.

생성자에서는 csv파일 한 줄을 가지고 분석에 필요한 데이터를 자신의 멤버변수에 저장한다.

7.항공 출발 지연 데이터 분석

연도별로 얼마나 많은 항공기가 출발이 지연되었는지 계산하는 프로그램.

항공 출발 지연 맵리듀스 입출력 데이터 타입

클래스 입출력 키                  값

\--------------------------------------------------------------

매퍼   입력   라인넘버             항공 운항 통계 데이터

​       출력   운항년도,운항월             출발 지연 건수

리듀서 입력   운항년도,운항월             출발 지연 건수

​       출력   운항년도 운항월      출발 지연 건수 합계

프로젝트 : hadoop_mapreduce

패키지 : AirlinePerformance

클래스 : AirlinePerformanceParser.java

매퍼 구현 : DepartureDelayCountMapper.java

리듀서 구현 : DelayCountReducer.java

드라이버 클래스 구현 : DepartureDelayCount.java

JAR : AirlinePerformanceDeparture.jar

입력폴더 : airline_input

출력폴더 : dep_delay_count

[hadoop@master jar]$ hadoop jar AirlinePerformanceDeparture.jar 
​       AirlinePerformance.ArrivalDelayCount airline_input arr_delay_count

[hadoop@master jar]$ hadoop fs -ls
Found 5 items
drwxr-xr-x  - hadoop supergroup     0 2017-03-25 12:58 airline_input
drwxr-xr-x  - hadoop supergroup     0 2017-03-26 09:51 dep_delay_count
[hadoop@master jar]$ hadoop fs -ls dep_delay_count
Found 2 items
-rw-r--r--  1 hadoop supergroup     0 2017-03-26 09:51 dep_delay_count/_SUCCESS
-rw-r--r--  1 hadoop supergroup    387 2017-03-26 09:51 dep_delay_count/part-r-00000
[hadoop@master jar]$ hadoop fs –cat dep_delay_count/part-r-00000

★ 항공 도착 지연 데이터 분석

클래스 입출력 키                  값

--------------------------------------------------------------

매퍼   입력   라인넘버             항공 운항 통계 데이터

​       출력   운항년도,운항월             도착 지연 건수

리듀서 입력   운항년도,운항월             도착 지연 건수

​       출력   운항년도 운항월      도착 지연 건수 합계

Package : AirlinePerformance

클래스 : AirlinePerformanceParser.java

매퍼 구현 : ArrivalDelayCountMapper.java

리듀서 구현 : DelayCountReducer.java

- 운항 년도,월을 묶어서 합계구하는 것은 동일

드라이버 클래스 구현 : ArrivalDelayCount.java

JAR : AirlinePerformanceArrival.jar

입력폴더 : airline_input

출력폴더 : arr_delay_count

[hadoop@master jar]$ hadoop jar AirlinePerformanceArrival.jar 

​       AirlinePerformance.ArrivalDelayCount airline_input arr_delay_count

[hadoop@master jar]$ hadoop fs -ls arr_delay_count
[hadoop@master jar]$ hadoop fs –cat arr_delay_count/part-r-00000
[hadoop@master jar]$ hadoop jar AirlinePerformanceArrival.jar 
​       AirlinePerformance.ArrivalDelayCount airline_input arr_delay_count
[hadoop@master jar]$ hadoop fs -ls arr_delay_count
[hadoop@master jar]$ hadoop fs –cat arr_delay_count/part-r-00000

[문제] 2013 ~ 2015년까지 범죄 조사

프로젝트 : hadoop_mapreduce

패키지 : CrimePerformance

클래스 : CrimePerformanceParser.java

매퍼 구현 : CrimeCountMapper.java

리듀서 구현 : CrimeCountReducer.java

드라이버 클래스 구현 : CrimeCount.java

JAR : CrimePerformanceType.jar

1.실습 데이터 내려받기

http://data.gov 접속 - Crime 검색

2014, 2015, 2016 년도 데이터를 다운로드 받는다.

2.다운로드 받은 파일을 workspace로 복사

3.첫째줄은 제목라인이다. 삭제하자.

[hadoop@master workspace]$ sed -e '1d' Crime_Data_2013.csv > Crime_Data_2013_new.csv
[hadoop@master workspace]$ sed -e '1d' Crime_Data_2014.csv > Crime_Data_2014_new.csv
[hadoop@master workspace]$ sed -e '1d' Crime_Data_2015.csv > Crime_Data_2015_new.csv

4.HDFS에 분석용 파일 업로드하기.

[hadoop@master workspace]$ hadoop fs -ls
[hadoop@master workspace]$ hadoop fs –mkdir crime_input
[hadoop@master workspace]$ hadoop fs –put Crime_Data_2013_new.csv crime_input
[hadoop@master workspace]$ hadoop fs –put Crime_Data_2014_new.csv crime_input
[hadoop@master workspace]$ hadoop fs –put Crime_Data_2015_new.csv crime_input
[hadoop@master workspace]$ hadoop fs -ls

5.소스작성

6.범죄 이벤트 데이터 분석

클래스 입출력 키                  값

--------------------------------------------------------------

매퍼   입력   라인넘버             범죄 이벤트 데이터

​       출력   년도,월,범죄유형      이벤트 건수

리듀서 입력   년도,월,범죄유형      이벤트 건수

​       출력   년도,월,범죄유형      이벤트 건수 합계

[hadoop@master jar]$ hadoop jar CrimePerformanceType.jar CrimePerformance.CrimeCount crime_input crime_output

DRUGS/ALCOHOL VIOLATIONS - 알콜위반
VEHICLE BREAK-IN/THEFT - 차량절도
VANDALISM - 공공기물 파손
ASSAULT - 폭행
DRUGS/ALCOHOL VIOLATIONS
VEHICLE BREAK-IN/THEFT
VEHICLE BREAK-IN/THEFT
ASSAULT 
VANDALISM
OTHER
BURGLARY - 절도

★ 항공 출발/도착 지연 데이터 분석_파라미터

항공 출발/도착 지연을 분석하기 위해 컴파일을 따로따로 하고, 실행도 따로 하였었다.

이것을 사용자 옵션을 넘겨받아서 workType 파라미터가 departure일때는 출발 지연을 분석하고, workType 파라미터가 arrival일때는 도착지연을 분석하도록 해보자.

패키지 : AirlinePerformanceWorkType

클래스 : AirlinePerformanceParser.java

매퍼 구현 : DelayCountMapper.java

리듀서 구현 (동일) : DelayCountReducer.java

드라이버 클래스 구현 : DelayCount.java

JAR : AirlinePerformanceWorkType.jar

[hadoop@master jar]$ hadoop jar AirlinePerformanceWorkType.jar AirlinePerformanceWorkType.DelayCount 
-D workType=departure airline_input departure_delay_count
[hadoop@master jar]$ hadoop fs -ls
[hadoop@master jar]$ hadoop fs -ls departure_delay_count
[hadoop@master jar]$ hadoop fs -ls -cat departure_delay_count/part-r-00000
[hadoop@master jar]$ hadoop jar AirlinePerformanceWorkType.jar AirlinePerformanceWorkType.DelayCount 
-D workType=arrival airline_input arrival_delay_count

1.GenericOptionsParser

하둡 콘솔명령어에서 입력한 옵션을 분석한다.

GenericOptionsParser는 사용자가 하둡 콘솔 명령에서 입력한 파라미터를 인식한다.

-D 를 이용하여 작업하면 파라미터별로 작업이 다르게 수행되도록 작성할 수 있다.

2.TOOL

Tool의 run메서드를 이용해서 하둡 실행시점에 입력한 파라미터를 읽어오고 적용할 수 있도록 작업할 수 있다.

3.ToolRunner

Tool 의 run메서드는 오직 ToolRunner에서만 호출 할 수 있다.

4.Mapper 작성

하둡을 실행 할 때 입력하는 파라미터 값에 따라 다르게 데이터를 매핑할 수 있도록 로직을 적용하는 작업을 수행

- 하둡 입력 시에 입력한 파라미터의 값을 추출 하는 작업

- 값에 따라 로직을 다르게 적용할 수 있도록 작업

5.Reducer 작성

출력된 값을 합치기만 하면 되므로 변경하지 않아도 된다.

6.Driver 작성

Configured와 Tool을 상속받도록 작성한다.

run메서드를 오버라이딩하여 실제 Driver에서 작업했던 모든 내용을 작업할 수 있도록 구현해야 한다.

만약 input과 output의 경로를 명령행 파라미터로 지정한다면 GenericOptionsParser 의 getRemainingArgs() 메서드를 이용하여 사용자 옵션에 지정한 파라미터를 제외한 값을 읽어들여야 한다.

★ 항공 출발/도착 지연 데이터 분석_카운터

카운터의 사용 : 맵리듀스 프레임워크의 맵에서 입력 파일을 한줄 한줄 읽을 때마다 조건문을 설정하여 조건에 맞을 경우 사용자 정의 카운터를 증가시킬 수 있다.

Ex. 항공기 데이터를 한줄 읽었는데, 조기 출발일 경우 사용자 정의 카운터 early_departure 1증가.

모든 파일을 읽은 후 카운터값 출력 해줌.

패키지 : AirlinePerformanceCounter

클래스 : AirlinePerformanceParser.java

매퍼 구현 : DelayCountMapperWithCounter.java

리듀서 구현 (동일) : DelayCountReducer.java

드라이버 클래스 구현 : DelayCountWithCounter.java

상수 : DelayCounters.java

JAR : AirlinePerformanceCounter.jar

[실행]

[hadoop@master jar]$ hadoop jar AirlinePerformanceCounter.jar AirlinePerformanceCounter.DelayCountWithCounter

-D workType=departure airline_input departure_delay_count_counter

[hadoop@master jar]$ hadoop jar AirlinePerformanceCounter.jar AirlinePerformanceCounter.DelayCountWithCounter

-D workType=arrival airline_input departure_delay_count_counter

★ 항공 출발/도착 지연 데이터 분석_멀티

위의 예제 까지는 도착과 출발의 결과를 얻기 위해 컴파일은 한번이지만 실행을 두번 하여야 했다.

이번에는 한번만 실행해서 출발과 도착을 한꺼번에 분석해서 파일을 2개로 나눠서 출력한다.

매퍼가 한줄의 입력 파일을 읽었을 때, 출발지연이면 매퍼의 출력 키에 D를 추가하고, 도착 지연이면 매퍼의 출력 키에 A를 추가한다. 그리고 리듀서에서는 A,D를 이용해서 도착과 출발을 구분하여 한꺼번에 처리한다.

패키지 : AirlinePerformanceMultiple

클래스 : AirlinePerformanceParser.java

매퍼 구현 : DelayCountMapperWithMultipleOutputs.java

리듀서 구현 : DelayCountReducerWithMultipleOutputs.java

드라이버 클래스 구현 : DelayCountWithMultipleOutputs.java

상수 : DelayCounters.java

JAR : AirlinePerformanceMultiple.jar

[실행]

[hadoop@master jar]$ hadoop jar AirlinePerformanceMultiple.jar AirlinePerformanceMultiple.DelayCountWithMultipleOutputs

airline_input  delay_count_multiple
 
[hadoop@master jar]$ hadoop fs -ls delay_count_multiple
Found 4 items
-rw-r--r--  1 hadoop supergroup    0 2016-02-26 15:08 delay_count_multiple/_SUCCESS
-rw-r--r--  1 hadoop supergroup   387 2016-02-26 15:08 delay_count_multiple/arrival-r-00000
-rw-r--r--   1 hadoop  supergroup    387 2016-02-26 15:08 delay_count_multiple/departure-r-00000
-rw-r--r--  1 hadoop supergroup    0 2016-02-26 15:08 delay_count_multiple/part-r-00000

[hadoop@master jar]$ hadoop fs –cat delay_count_multiple/departure-r-00000
[hadoop@master jar]$ hadoop fs –cat delay_count_multiple/arrival-r-00000

[문제] 2013 ~ 2015년까지 범죄 조사

범죄유형(CRIME_TYPE), 범죄현장(PREMISE_TYPE), 위치(BLOCK_ADDRESS)에 따른 각각의 파일을 MultiOutputs를 이용하여 한번에 구하시오

프로젝트 : hadoop_mapreduce

패키지 : CrimePerformanceMultiple

클래스 : CrimePerformanceParser.java

매퍼 구현 : CrimeCountMapperMultiple.java

리듀서 구현 : CrimeCountReducerMultiple.java

드라이버 클래스 구현 : CrimeCountMultiple.java

JAR : CrimePerformanceMultiple.jar

[hadoop@master jar]$ hadoop jar CrimePerformanceMultiple.jar CrimePerformanceMultiple.CrimeCountMultiple 

crime_input crime_output_multiple

[hadoop@master jar]$ hadoop fs -ls crime_output_multiple
Found 5 items
-rw-r--r--  1 hadoop supergroup     0 2017-02-28 20:13 crime_output_multiple/_SUCCESS

-rw-r--r--  1 hadoop supergroup  9433906 2017-02-28 20:13 crime_output_multiple/blockAddress-r-00000

-rw-r--r--  1 hadoop supergroup   12881 2017-02-28 20:13 crime_output_multiple/crimeType-r-00000

-rw-r--r--  1 hadoop supergroup     0 2017-02-28 20:13 crime_output_multiple/part-r-00000

-rw-r--r--  1 hadoop supergroup   50775 2017-02-28 20:13 crime_output_multiple/premiseType-r-00000

[hadoop@master jar]$ hadoop fs -cat crime_output_multiple/crimeType-r-00000

[hadoop@master jar]$ hadoop fs -cat crime_output_multiple/premiseType-r-00000

[hadoop@master jar]$ hadoop fs -cat crime_output_multiple/blockAddress-r-00000

★ MapReduce 정렬 기법

Hadoop 정렬

하둡의 맵리듀스는 기본적으로 입력 데이터의 키를 기준으로 정렬되기 때문에 하나의 리듀스 태스크만 실행되게 한다면 정렬을 쉽게 해결 할 수 있습니다

하지만 여러 데이터노드가 구성된 상황에서는 하나의 리듀스 태스크만을 실행하지 않을 것입니다. 왜냐면 분산환경의 장점을 사용하지 않는 것이기 때문입니다.

따라서 우리는 분산환경에서의 여러 데이터 노드를 이용하면서 정렬을 할 수 있어야 한다.

우린 이것을 가능하게 하기 위해 보조 정렬(Secondary Sort), 부분 정렬(Partial Sort), 전체 정렬(Total Sort)을 사용할 수 있다.

보조 정렬(Secondary Sort)

보조 정렬은 키의 값들을 그룹핑하고, 그룹핑된 레코드에 순서를 부여하는 방식

사용자 정의 파티셔너와 GroupingComparator를 이용할 수 있습니다.

1.기존 키의 값들을 조합한 복합키(Composite Key)를 정의합니다.

이때 키의 값 중에서 어떤 키를 그룹핑 키로 사용할지 결정합니다.

복합키는 기존의 키값을 조합한 일종의 키 집합

2.복합키의 레코드를 정렬하기 위한 비교기(Comparator)를 정의합니다.

복합키 비교기는 두개의 복합키를 비교하여 정렬 순서를 정함

3.그룹핑 키를 파티셔닝할 파티셔너(Partitioner)를 정의합니다.

파티셔너는 맵 태스크의 출력 데이터를 리듀스 태스크의 입력 데이터로 보낼지 결정

파티셔닝된 데이터는 맵 태스크의 출력 데이터의 키의 값에 따라 결정

4.그룹핑 키를 정렬하기 위한 비교기(Comparator)를 정의합니다.

리듀서는 그룹키 비교기를 사용해 그룹핑 되어있는 데이터(같은 연도에 해당하는 모든 데이터)를 하나의 Reducer 그룹에서 처리

[실습] 보조 정렬(Secondary Sort)

먼저 실습하기 전에 간단하게 Writable에 대한 자바를 공부한다

Project : hadoop_sort

Package : exam

Class : Person

PersonMain

Project : hadoop_sort

Package : AirlinePerformanceSecondarySort

AirlinePerformanceParser.java

DelayCounters.java

DateKey.java

DateKeyComparator.java

GroupKeyPartitioner.java

GroupKeyComparator.java

DelayCountMapperWithDateKey.java

DelayCountReducerWithDateKey.java

DelayCountWithDateKey.java

JAR : AirlinePerformanceSecondarySort.jar

[hadoop@master jar]$ hadoop jar AirlinePerformanceSecondarySort.jar 
AirlinePerformanceSecondarySort.DelayCountWithDateKey
airline_input delay_count_sort

...중략
 
       Map-Reduce Framework

              Map input records=11555122

              Map output records=11273069

              Map output bytes=202915242

              Map output materialized bytes=225461434

              Input split bytes=1098

              Combine input records=0

              Combine output records=0

              Reduce input groups=6 – 87년 출발,지연

                     88년 출발,지연

                     89년 출발, 지연 => 총 6개

              Reduce shuffle bytes=225461434

              Reduce input records=11273069

              Reduce output records=0

              Spilled Records=22546138

              Shuffled Maps =9

              Failed Shuffles=0

              Merged Map outputs=9

              GC time elapsed (ms)=16648

              CPU time spent (ms)=78070

              Physical memory (bytes) snapshot=2027245568

              Virtual memory (bytes) snapshot=20612046848

              Total committed heap usage (bytes)=1439666176

       AirlinePerformanceSecondarySort.DelayCounters

              early_arrival=4380215

              early_departure=1984103

              not_available_arrival=177103

              not_available_departure=144013

              scheduled_arrival=567687

              scheduled_departure=4584054

부분 정렬(Partial Sort)

부분 정렬은 매퍼의 출력 데이터를 맵파일(MapFile)로 변경해서 데이터를 검색하는 방법

부분 정렬은 정렬할 파일을 맵 단계에서 시퀀스 파일로 출력을 해주고, 출력된 시퀀스 파일을 맵파일로 변경한 후 맵파일에서 원하는 데이터를 검색해내는 방법입니다.

맵파일이 여러 개가 만들어져 있더라도, 각각의 맵파일에는 키에 맞는 데이터들만 저장되어 있어서 데이터 검색이 가능한 것입니다.

1.입력 데이터를 시퀀스 파일로 생성합니다.

- 매퍼는 입력 데이터를 연산하지 않기 때문에 리듀서가 필요 없음

2.시퀀스 파일을 맵파일로 변경합니다.

- 맵파일은 키값을 검색할 수 있게 색인과 함께 정렬된 시퀀스 파일

- 맵파일은 물리적으로 색인이 저장된 index 파일과 데이터 내용이 저장돼 있는 data 파일로 구성

3.맵파일에서 데이터를 검색합니다.

- 검색의 키는 파티셔너

- 파티셔닝된 데이터는 맵 태스크의 출력 데이터의 키의 값에 따라 결정

데이터를 다루다 보면 전체 정렬이 필요한 경우도 있지만 지금처럼 특정 키에 해당하는 데이터만 검색해서 사용해야 할 경우 부분 정렬을 활용

[실습] 부분 정렬(Partial Sort)

패키지 : AirlinePerformancePartialSort

AirlinePerformanceParser.java

SequenceFileCreator.java

MapFileCreator.java

SearchValueList.java

JAR : AirlinePerformancePartialSort.jar

1.시퀀스 파일

1988년 통계 데이터로 맵리듀스 잡을 실행

[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar
AirlinePerformancePartialSort.SequenceFileCreator
airline_input/1988_new.csv 1988_sequence
[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar
AirlinePerformancePartialSort.SequenceFileCreator

airline_input airline_sequence ⟶ airline_input폴더에 87,88,89년도의 모두 통합해서 sequence파일을 만든다

[hadoop@master jar]$ hadoop fs -ls 1988_sequence
Found 5 items
-rw-r--r--  1 hadoop supergroup     0 2017-02-02 12:07 1988_sequence/_SUCCESS
-rw-r--r--  1 hadoop supergroup  19913688 2017-02-02 12:07 1988_sequence/part-00000
-rw-r--r--  1 hadoop supergroup  19122032 2017-02-02 12:07 1988_sequence/part-00001
-rw-r--r--  1 hadoop supergroup  19117352 2017-02-02 12:07 1988_sequence/part-00002
-rw-r--r--  1 hadoop supergroup  14328075 2017-02-02 12:07 1988_sequence/part-00003
[hadoop@master jar]$ hadoop fs -cat 1988_sequence/part-00000 | more
[hadoop@master jar]$ hadoop fs -text 1988_sequence/part-00000 | more

2.맵 파일

[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar

AirlinePerformancePartialSort.MapFileCreator 1988_sequence 1988_mapfile

[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar

AirlinePerformancePartialSort.MapFileCreator airline_sequence airline_mapfile

[hadoop@master jar]$ hadoop fs -ls 1988_mapfile
Found 2 items
-rw-r--r--  1 hadoop supergroup     0 2017-03-06 20:49 1988_mapfile/_SUCCESS
drwxr-xr-x  - hadoop supergroup     0 2017-03-06 20:49 1988_mapfile/part-00000
 
[hadoop@master jar]$ hadoop fs -ls 1988_mapfile/part-00000
Found 2 items
-rw-r--r--  1 hadoop supergroup  70574032 2017-03-06 20:49 1988_mapfile/part-00000/data
-rw-r--r--  1 hadoop supergroup    3349 2017-03-06 20:49 1988_mapfile/part-00000/index

맵파일의 규격에 맞게 data와 index파일이 생성되어 있다.

[hadoop@master jar]$ hadoop fs -text 1988_mapfile/part-00000/index | more

엉망으로 보였던 데이터가 정렬된 것을 확인할 수 있다

[hadoop@master jar]$ hadoop fs -text 1988_mapfile/part-00000/data | more
[hadoop@master jar]$ hadoop fs -text 1988_mapfile/part-00000/data | head -10

3.검색 프로그램

검색프로그램 실행 전 주의해야 할 사항은 현재 mapfile 안에 있는 로그폴더를 모두 삭제해야 해야 한다

MapFileOutputFormat 클래스가 mapfile폴더 안에 맵리듀스 잡으로부터 생성된 로그파일폴더를 체크해서 맵파일이 없다는 오류를 발생시키기 떄문입니다.

[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar

AirlinePerformancePartialSort.SearchValueList 1988_mapfile 273

Exception in thread "main" java.io.FileNotFoundException: File does not exist: hdfs://master:9000/user/hadoop/1988_mapfile/_SUCCESS/data
...
[hadoop@master jar]$ hadoop fs -ls 1988_mapfile

Found 2 items
-rw-r--r--  1 hadoop supergroup     0 2017-03-06 20:49 1988_mapfile/_SUCCESS
drwxr-xr-x  - hadoop supergroup     0 2017-03-06 20:49 1988_mapfile/part-00000

[hadoop@master jar]$ hadoop fs -rm -r 1988_mapfile/_SUCCESS
[hadoop@master jar]$ hadoop fs -ls 1988_mapfile
Found 1 items
drwxr-xr-x  - hadoop supergroup     0 2017-03-06 20:49 1988_mapfile/part-00000

[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar

AirlinePerformancePartialSort.SearchValueList 1988_mapfile 273 | more

[hadoop@master jar]$ hadoop jar AirlinePerformancePartialSort.jar

AirlinePerformancePartialSort.SearchValueList 1988_mapfile 20 | more

...

The requested key was not found

전체 정렬(Total Sort)

모든 맵리듀스 잡은 입력데이터의 키를 기준으로 정렬하기 때문에, 하나의 파티션으로 손쉽게 데이터를 정렬할 수 있다.

그러나 단일 파티션을 사용할 경우 크기가 큰 데이터를 정렬하게 되면 문제가 발생하게 됩니다. 리듀스 태스크를 실행하지 않는 데이터노드는 가동되지 않고, 리듀스 태스크가 실행되는 데이터노드만 부하가 집중 될 것이기 때문입니다.

분산 처리의 장점을 살리면서 전체 정렬을 하려면 다음과 같은 순서로 정렬을 진행

1.입력 데이터를 샘플링해서 데이터의 분포도를 조사합니다.

- 입력 데이터에서 특정 개수의 데이터를 추출해서 키와 데이터 건수를 샘플링

2.데이터의 분포도에 맞게 파티션 정보를 미리 생성합니다.

- 샘플링 결과를 기준으로 파티션을 생성(샘플러가 제공하는 키의 정보를 설정)

- 각 task가 파티션 정보를 참조할 수 있도록 분산 캐시에 파티션 정보를 등록

- TotalOrderPartitioner로 파티션개수와 파티션에 저장할 데이터범위를 설정할 수 있다

- 시퀀스 파일로 출력(TotalOderPartitioner는 시퀀스 파일에 최적화)

3.리듀스 잡은 TotalOderPartitioner가 생성한 파티션에 출력 데이터를 생성합니다.

4.각 출력 데이터를 병합 합니다.

※ InputSampler 종류

SplitSampler : 입력 스플릿에서 첫 번째 레코드의 키를 수집

IntervalSampler : 입력 스플릿에서 일정한 간격으로 키를 수집

RandomSampler : 일정 스플릿 건수에서 일정 확률로 키를 수집함. 이때 스플릿 건수와 확률은 사용자가 설정하며, 샘플링한 데이터 건수도 임의로 설정할 수 있다.

각 파티션은 키 순서대로 정렬돼 있으며, 이 파일들을 합치면 전체 정렬한 효과를 보게 됩니다.

이 시점에 부분 정렬을 하지 않고 전체 정렬을 하는게 더 편하다고 생각할 수 도 있습니다.

물론 두 정렬은 모두 부분 정렬을 하지만 가장 큰 차이점은 검색입니다.

정렬된 결과에서 데이터를 검색할 필요가 있다면 부분 정렬을, 단순히 정렬된 전체 데이터만 필요하다면 전체 정렬을 이용하면 됩니다.

[hadoop] hadoop_2

Hadoop_2

🐯

Hadoop_2

Share this post