이번 포스팅에서는 RDD 객체를 DataFrame Dataset으로 convert하고, Spark SQL로 데이터를 전처리한 실습내용에 대해서 정리해보려고 한다.

RDD => DataFrame

RDD를 DataFrame으로 변경함으로써 Spark SQL을 통해 데이터를 쉽게 가공할 수 있다.
아래는 샘플 코드로, RDD데이터를 spark.createDataFrame을 통해서 DataFrame으로 convert하고, convert된 DataFrame 객체를 활용해서 Spark SQL을 활용해서 groupBy, avg, count, join, orderBy, take, 등을 수행한다.

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

from pyspark.sql import SparkSession
from pyspark.sql import Row
from pysql.sql import functions

# Convert RDD to DataFrame
df = spark.createDataFrame([RDD DATA])

# movieID 기준으로 grouping하고, rating에 대한 평균값 column을 표시한다.
averageRatings = df.groupBy("movieID").avg("rating")

# 각 movieID에 대해서 rating을 count한다.
counts = df.groupBy("movieID").count()

# movieID, avg(rating), count 세 개의 column을 join한다.
averagesAndCounts = counts.join(averageRatings, "movieID")

# top 10 results를 출력한다.
topTen = averagesAndCounts.orderBy("avg(rating)").take(10)

# Stop the session
spark.stop()

이번 포스팅은 Spark에 대해서 정리를 해보려고 한다. 데이터 파이프라인 실습에서 이미 Spark interpreter를 사용해서 데이터 전처리를 하고 있다. 그런데 아직 정확히 Spark에 대한 개념적 정의가 되지 않았기 때문에 이 부분에 대해서 정리를 하고 넘어가려고 한다.

Spark ?

Spark는 거대한 양의 데이터를 합리적인 방법으로 처리하고, 더 나아가 ML이나 Graph 분석 그리고 Data streaming 등의 멋진 기능을 포함하고 있는 친구이다.
이전에 데이터 파이프라인 구축 실습에서 데이터 batch 처리(ETL)나 실시간 스트리밍 데이터를 처리할때도 모두 이 Spark라는 친구를 사용한다고 했으니, 정말 대단한 친구임에는 틀림이 없다.
이처럼 Spark의 역량과 속도는 엄청나며, Java나 Scala, Python과 같은 실제 프로그래밍 언어를 사용해서 스크립트를 작성할 수 있는 유연성을 제공하고, 복잡한 데이터를 조작이나 변형, 분석할 수 있는 능력자이다.

이번 포스팅에서는 앞서 배운 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를 창의적으로 조합하면, 놀랄만큼 다양한 분석문제를 해결할 수 있다.

이번 포스팅에서는 sample 데이터를 활용해서 HDFS를 실제로 다뤄보는 실습을 해본다.

Ambari

Ambari는 Hadoop Eco System에서 구조상 최 상위에 위치해있는 어플리케이션으로, Hadoop과 관려된 서비스들 전체를 모니터링할 수 있는 웹 인터페이스 환경을 제공한다.

웹 인터페이스를 사용해서 HDFS 다뤄보기

HTTP 인터페이스로 간단하게 HDFS에 파일을 업로드, 로컬 디스크에 다운로드, HDFS에 업로드한 파일들을 삭제해보았다.

HDFS에 파일을 업로드할때에는 단순히 파일을 업로드 되는 것처럼 보이지만, 실제로는 파일을 올릴때 HDFS 내부의 Name Node에 어느 Data Node에 저장을 할지 확인하는 절차를 통해 클러스터내의 특정 Data Node에 저장을 하고, 정상적으로 저장이 완료되면 Name Node에 결과를 알려줘서 Name Node가 관리하고 있는 저장 데이터와 관련된 메타 데이터 테이블에 새롭게 저장된 데이터 정보를 추가하게 된다.
(내부적으로 이러한 일련의 과정을 거치고 있다는 것을 인지한 상태에서 계속해서 학습을 이어가도록 하자)

이번 포스팅에서는 HDFS의 내부구성에 대해서 알아보고, 대용량의 파일을 어떤식으로 처리하는지에 대해서 한 번 정리해보겠다.

개괄적으로 HDFS에서 대용량의 파일을 처리하는 방법에 대해서 간단히 알아보면 대용량의 파일을 여러개의 블록 단위로 쪼개서 클러스터를 구성하고있는 각 노드에 저장을 한다.
각 노드는 같은 데이터 블록 정보를 저장하고 있기 때문에 노드 중에 한 개에 문제가 생겨도 복원을 할 수 있다.
(Single point failure issue 대응)

이제 한 번 STEP BY STEP으로 HDFS(Hadoop Distribution File System)에 대해서 알아보자.

STEP1 BIG DATA의 등장과 블록단위로 쪼개고 저장하기

자, 엄청나게 큰 데이터가 주어졌다고 가정하자.
HDFS는 이러한 빅데이터를 전체 클러스터에 분산해서 안정적으로 저장해서 어플리케이션이 그 데이터를 신속하게 액세스해 분석할 수 있도록 돕는다.

좀 더 구체적으로 말하면 HDFS는 대용량 파일들을 다루기 위해서 만들어졌다. 작은 데이터는 누구든 쉽게 다룰 수 있지만, HDFS는 이런 대용량의 파일들을 작은 조각으로 쪼개서 클러스터 전체에 걸쳐 분산시키는데 최적화되어있다.
그래서 분산 파일 시스템이라고 하는 것이다.