이번 포스팅에서는 앞서 배운 Pig Latin script 작성방법을 기반으로 Ambari를 통해서 Hadoop 클러스터에서 Pig script를 실행해보고 결과값을 확인해볼 것이다.

Ambari에서의 Pig script 실행

Hadoop stack을 사용해서 Pig Latin script를 실행해보았다.
결과는 Ambari에서 결과, Logs에 대한 기록, 작성한 Script Details 정보를 각 카테고리별로 분류해서 확인할 수 있었다.

이번 포스팅에서는 구조상으로 보면, MapReduce의 상위에 위치해있는 Pig라는 친구를 사용해서 실습을 해보도록 하겠다. 실습은 Ambari를 사용해서 해보도록 한다.

Ambari에서 관리자 계정 접근 활성화

1
2
3
4

$su root
# switch to root account
$ambari-admin-password-reset
# type the password for admin account

Pig ?

Ambari 웹 브라우저의 화면 우측 상단에 격자 무늬 아이콘을 클릭하면, Pig View 메뉴를 확인할 수 있는데, 이곳에 Pig Scripts 추가하고 실행해볼 수 있다.

여지까지 Hadoop에 대해서 학습하면서 Hadoop의 가장 핵심기술인 MapReduce에 대해서 중점적으로 배웠다. MapReduce는 Hadoop의 시작을 함께 한 기술이긴 하지만 오래된 데이터 처리 기술이고, 어렵다. 이러한 어려운 데이터 처리 기술을 Pig는 좀 더 쉽게 할 수 있도록 도와준다.

이번 포스팅에서는 MapReduce 실습 도전과제에 대해 직접 해본 내용을 정리해보려고 한다.

이전 포스팅에서는 영화 정보 데이터가 주어졌을때, 각 평점을 기준으로 몇 편의 영화가 분포되어있는지 확인하는 MapReduce 코드를 작성해보았다.

이번 Self-Challenge 과제는 총 2개이며,
첫 번째 챌린지 과제는 영화별 인기순위를 분류하기 위한 MapReduce 작성하는 것이다.
두 번째 챌린지 과제는 평가 횟수를 기준으로 영화를 정렬하는 MapReduce를 작성하는 것이다.

영화별 인기순위 분류하기 위한 MapReduce 작성

영화 인기순위는 영화 평가 횟수를 기준으로 보았을 때, 시청 횟수의 대용물로 평가가 될 수 있기 때문에 movieID를 기준으로 mapper를 작성해주고, reducer에서 해당 영화 평가 횟수를 종합해주면 된다는 결론이 나온다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from mrjob.job import MRJob
from mrjob.step import MRStep

class NumOfEvaluateBreakdown(MRJob):
  def steps(self):
    return [
      MRStep(mapper=self.mapper_get_num_of_evaluate,
            reducer=self.reducer_count_num_of_evaluate)
    ]
  def mapper_get_num_of_evaluate(self, _, line):
    (userID, movieID, rating, timestamp) = line.split('\t')
    yield movieID, 1

  def reducer_count_num_of_evaluate(self, key, values):
    yield key, sum(values)

  if __name__ == '__main__':
    NumOfEvaluateBreakdown.run()

이번 포스팅에서는 이전에 작성했던 Python으로 작성한 MapReduce 코드를 HDP 2.6.5 환경에서 구동시켜볼 것이다. Hadoop을 사용하지 않고, local에 copy된 dataset을 가지고 로컬에서 실행해보고, Hadoop을 기반으로 실행해보는 두 가지 방법으로 실습을 해볼 것이다.

HDP2.6.5에서 MapReduce 실습환경 구축

[STEP1] 가장 먼저 VirtualBox에 올린 HDP2.6.5 OS Image를 구동시킨다.

[STEP2] Putty를 사용해서 가상 OS환경에 접속한다.

[Putty host/port configuration]

1
2

host: maria_dev@127.0.0.1
port: 2222

[STEP3] PIP 설치를 위한 SETUP

1
2
3
4
5
6
7
8
9
10
11
12
13

# root 계정으로 switch

$su root # root account initial password : hadoop

$yum-config-manager --save --setopt=HDP-SOLR-2.6-100.skip_if_unavailable=true # HDP Solar라는 저장소를 무시하고 설치하기 위한 configuration

$yum install https://repo.ius.io/ius-release-el7.rpm https://dl.fedoraproject.org/pub/epel/epel-release-latest-7.noarch.rpm # IUS package를 수동으로 설치

$yum install python-pip # python-pip 설치

$pip install pathlib # pathlib 설치 *설치를 안하면 MRJob이 정상적으로 설치가 안됨

$pip install mrjob==0.7.4 #설치한 pip를 이용해서 MRJob 설치

이번 포스팅에서는 Hadoop의 생태계에서 핵심이 되는 MapReduce의 세부동작에 대해서 정리해보려고 한다.

MapReduce에 동작에 대해서 이전 포스팅에서 정리를 했듯이 Mapper와 Reducer가 하는 일은 그렇게 복잡해보이지 않는다. 다만 Hadoop의 Cluster내에서 MapReducer가 동작하는 방식이 복잡하기 때문에 이 부분에 대해서 정리가 필요하다.

이전 포스팅에서 다뤘던 내용은 입력받은 데이터를 Mapper가 Key-Value 쌍으로 데이터를 Transformation해주고, 그 결과를 MapReduce가 자체적으로 셔플과 정렬(Shuffling and Sorting)해준다.
이후에 Reducer는 구조화된 정보를 전달받아서 최종 출력물을 생산해내는 역할을 한다.

만약에 정말로 큰 데이터 세트를 가진 클러스터를 운영한다면, 아마 처리 과정을 여러 컴퓨터에 배분하거나 각 노드에서 여러 작업에 걸쳐 진행을 할 것이다.
MapReduce 작업을 세 개의 노드에 나눠서 Mapping작업을 한다고 가정한다면, 데이터에서 몇 줄은 첫 번째 노드에 보내서 처리하고, 나머지 데이터는 나머지 노드에 배분해서 처리할 것이다.
입력 데이터를 여러 파티션에 끼워 맞추고, 각 파티션에 작업 할당을 하게 되는 것이다.
바로 여기서 다른 파티션에 있는 데이터는 신경쓰지 않아도 되기 때문에 작업의 병렬화가 가능해진다.
최종적으로 각 각 작업 배분한 컴퓨터에서 작업이 끝나면 Hadoop은 정보를 잘 받아오고 마무리가 된다.

[MapReduce procedure 반복 숙지]

[STEP1 - Mapping]
클러스터의 각 노드로 대량의 데이터를 작은 블록으로 배분함으로써 key:value 쌍으로 전환하는 Mapping 처리과정을 배분할 수 있다.

[STEP2 - Shuffling and Sorting]
그 다음으로는 Shuffle & Sort작업인데, 앞서 key:value 쌍으로 Mapping하는 과정에서 복수 개의 key를 갖는 데이터가 생기기 때문에 이 과정을 통해서 같은 키 값끼리 모아서 Reducer로 보낼 데이터로 가공하게 된다. 이 과정은 MapReduce가 대신 수행을 하게 된다. 단순히 네트워크상에서 데이터를 주고 받는 형태가 아닌, 모든 정보를 merge sort하는 형태로 처리한다.

[STEP3 - Reducing]
최종적으로 Reducer 단계에서는 데이터를 Key 값을 기준으로 정렬된 세트를 Reducing하게 된다.

다시 정리하자면, Mapping단계에서는 입력된 데이터를 작은 block으로 쪼개서 클러스터 내의 여러 노드로 분산시키고, 셔플과 정렬 작업 이후에는 각 노드가 각 각의 세트를 담당하여 reducing하게 된다. 클러스터 내의 각 노드에서 reducing 작업까지 완료된 데이터들은 Hadoop에 의해 최종적으로 수집되어 마무리된다.

이번 포스팅에서는 MapReduce의 정의에 대해서 알아보고, 어떻게 작동을 하는지 그리고 분산처리하는 방법과 간단한 MapReduce 연습문제를 통해 데이터를 분류해보도록 하겠다.

MapReduce?

이전에 Hadoop의 Core System에 대해서 전반적으로 알아볼때 이미 학습했지만, MapReduce는 HDFS -> YARN -> MapReduce 순으로 구조화되어있다. 즉, MapReduce는 HDFS, YARN과 함께 제공되는 Hadoop의 핵심기술이다.
Hadoop을 이해하기 위해서는 MapReduce를 잘 이해하는 것이 필수이다.

MapReduce는 Hadoop에 내제된 기능으로, 클러스터에 데이터의 처리를 분배하는 역할을 한다.데이터를 파티션으로 나눠서 클러스터 전반에 걸쳐서 병렬 처리되도록 한다.
이러한 일련의 과정에서 관리를 하고 실패하였을때 감당하는 역할을 한다.

Data Mapping -> Mapper -> Data Transformation

데이터를 변형(Transformation)하는 과정으로, 데이터가 한 줄씩 들어오면, Mapper는 해당 데이터를 변형시키게 된다. 들어오는 데이터에서 필요한 정보를 추출하고, 이해할 수 있는 구조로 정리를 해서 해결하고자하는 문제를 해결한다.

데이터를 집계하고자 하는 중요한 데이터를 키-값의 형태로 연관시켜서 구조화시킨다. (키:집계하고자하는 항목)
input line -> (mapper) -> Data 추출 및 구조화해서 한 줄로 출력

Mapper의 처리 순서는 아래와 같다.

INPUT DATA -> Mapper -> K1:V, K2:V, K3:V, K4:V, K5:V ….

Data Reducing -> Reducer -> Data Aggregation

Reducer는 데이터를 집계한다. Mapper를 통해 key:value의 형태로 변형된 데이터가 shuffle & sort 과정을 거쳐서 Reducer는 얻고자 하는 결과 데이터를 기준으로 key:value에서 value 데이터를 집계한다.

[MapReducer를 통한 문제해결]

Mapper와 Reducer를 창의적으로 조합하면, 놀랄만큼 다양한 분석문제를 해결할 수 있다.