.

Data_Engineering_TIL(20211212)

[학습자료]

‘형준킴 염창동형준킴’님의 아래에 블로그를 읽고 공부한 내용입니다.

1) ‘형준킴 염창동형준킴’님 블로그 “갈아먹는 BigData[2] HDFS(하둡 분산 파일 시스템)”

** URL : https://yeomko.tistory.com/38?category=878347

2) ‘형준킴 염창동형준킴’님 블로그 “갈아먹는 BigData [1] MapReduce 이해하기”

** URL : https://yeomko.tistory.com/31?category=878347

[학습내용]

• 맵리듀스 핵심개념

1) 맵 리듀스는 구글 내부에서 크롤링 된 문서, 로그 등 방대한 양의 raw data를 분석하는 과정에서 느낀 불편함에서 개발을 하게 되었음

2) 프로그램 로직 자체는 단순한데 입력 데이터의 크기가 워낙 방해서서 연산을 하나의 물리 머신에서 수행할 수가 없었기 때문에 이 거대한 input 데이터를 쪼개어 수많은 머신들에게 분산시켜서 로직을 수행한 다음 결과를 하나로 합치자는 것이 핵심 아이디어임

3) MapReduce 프레임워크에서 개발자가 코드를 작성하는 부분은 map과 reduce 두 가지 함수임. map은 전체 데이터를 쪼갠 청크에 대해서 실제로 수행할 로직임. 텍스트 데이터라면 텍스트를 파싱해서 단어의 개수를 세는 것이 대표적인 예시임. reduce는 분산되어 처리된 결과 값들을 다시 하나로 합쳐주는 과정이며, 이 역시 분산된 머신들에서 병렬적으로 수행됨

• 기본적인 맵리듀스 프로세스

STEP 1) split(데이터 쪼개기) : 크기가 큰 인풋 데이터를 작은 단위의 청크단위로 쪼개서 HDFS에 저장

STEP 2) map(데이터 처리하기) : 잘개 쪼개어진 파일을 인풋으로 받아서 데이터를 분석하는 로직을 수행

STEP 3) Reduce(처리된 데이터 합치기) : 처리된 데이터를 다시 합침

• 맵리듀스 분산처리 도식화

(1) fork

인풋 파일은 미리 쪼개져서 이미 HDFS에 저장된 상태라고 가정하자. 사용자가 map과 reduce 함수를 정의한 프로그램을 실행시켜서 프로세스를 실행하면 이 프로그램은 마스터 노드, map을 실행할 워커 노드(= Mapper 노드), reduce를 실행할 워커 노드(= Reducer 노드)들에 복사됩니다

(2) assign map and reduce

마스터 노드는 Mapper워커들에게 mapping 역할을, reducer 노드들에게는 reduce  역할을 수행하라고 지정해줌. 이 때, mapper 노드의 개수나 reducer 노드의 개수는 사용자가 설정이 가능함.

(3)(4) read, map, local write

Mapper 노드들은 쪼개어진 데이터 청크를 HDFS로부터 읽어옴. 그 다음 이 데이터에 Map 함수를 실행하여 K:V 형태의 Intermediate 데이터를 생성하고 이를 자기 자신의 로컬 디스크에 저장함. 그 다음 Mapper 노드는 마스터 노드에게 Mapping 작업을 모두 완료하였다고 알려줌.

(5)(6) reduce, write

그러면 마스터 노드는 리듀서 노드들에게 reduce를 시작하라고 명령을 내려줌. reducer 노드들은 mapper 노드들의 디스크 공간에 저장되어 있는 Intermediate 데이터를 읽어옴. 그리고 reduce 함수를 실행하여 최종 결과물을 산출한 다음 파일 형태로 데이터를 출력하게 됨.

• Word Counting Example

가장 유명한 예시인 워드 카운팅을 통해서 MapReduce를 좀더 알아보자. 아래 그림과 같이 3대의 Mapper와 4대의 Reducer 노드로 이루어진 클러스터에서 워드 카운팅을 수행하는 상황이라고 가정하자.

STEP 1) Splitting, Mapping

먼저input 텍스트 파일을 쪼개어 HDFS에 저장합니다. 쪼개어진 텍스트 데이터는 3대의 Mapper노드에 각각 전달됨. 그리고 각 노드들은 Map 함수를 실행하여 입력으로 들어온 텍스트 데이터를 파싱하여 Dear:1, Bear:1, River:1 과 같은 Intermediate Key Value를 생성하게 됨.

STEP 2) Shuffling

Mapping으로 생성된 Intermediate Key Value 값을 Reducer 노드에 입력으로 전달하기 이전의 과정을 통틀어서 Shuffling 이라고 한다.

1) Shuffling이 필요한 이유

우리가 MapReduce를 통해서 알고 싶은 것은 전체 텍스트에서 특정 단어가 얼마나 자주 등장했는지 인데 이를 위해서 전체 텍스트 데이터를 3조각으로 나누었고, 이를 Mapper 노드들에 분산 시켜 워드 카운팅을 진행했다. 문제는 이렇게 생성된 Intermediate Key Value 들에서 노드는 서로 다르지만 같은 키 값을 지닌 쌍들이 있다는 것이다.

예를들어 Dear라는 단어는 1번 Mapper와 3번 Mapper의 Intermediate Key Value에는 Dear:1이라는 값이 있다. 이를 적절히 Reduce 해주기 위해서는 이 값들이 모두 동일한 Reducer에게 전달이 되어야만한다. 즉, 동일한 키 값을 지닌 Intermediate Key Value Pairs는 동일한 Reducer에게 전달되어야 하며, 이 매핑 과정을 Shuffling이라고 함.

2) Shuffling 프로세스

map 함수를 거쳐서 생성된 intermediate key value pairs는 먼저 메모리 상의 버퍼에 저장됨. 그 다음 partitioning이라는 과정을 거치게 됨. partitioning이란 해당 키가 처리되어야 하는 reduce task를 기준으로 데이터를 구분해줌. 보통 키 값을 해쉬하여 파티션을 결정하게됨. 하둡에서는 기본적으로 키 값을 리듀서의 개수로 해쉬하는 방식을 사용함.

예를들어서 key mod R(num of reducers)) 이런식으로 해주면 각각의 키 값들이 어느 리듀서에서 처리되어야 하는 지를 구분할 수 있음. 각각의 파티션은 다시 키 값을 기준으로 소팅을 거쳐서 Mapper 노드의 local disk 공간에 저장된다.

참고로 리듀서의 개수를 기준으로 키 값을 단순히 해쉬하게 되면 빈번하게 등장하는 단어 처럼 특정 reduce 작업의 부하가 높아질 우려가 있기때문에 하둡은 직접 파티셔닝 로직을 작성할 수 있는 기능을 제공함.

이렇게 파티셔닝 된 데이터는 이제 Combining이라는 과정을 거친다. Combining이란 파티션 내의 Key Value 데이터들에 대해서 리듀스 작업을 진행하여 리듀서 노드들의 부하를 줄여주는 작업을 말한다. 워드 카운팅의 예시를 들어보면 아래와 같습니다.

컴파이닝 과정을 거친 데이터는이제 Mapper 노드의 local disk 공간에서 저장되고 Mapping 과정을 완료했다는 정보와 파티셔닝 된 데이터의 위치 정보를 마스터 노드에게 알려주게 된다. 마스터 노드는 Mapper들로부터 전달받은 정보들을 취합하여 Reducer 노드들에게 명령을 내린다. Reducer 노드들은 이제 각각의 Mapper 노드들에서 자신에게 할당된 파티션에 해당하는 데이터만 읽어와 reduce 작업을 진행함으로써 셔플링 과정이 마무리 됨.

3) Shuffling 프로세스 요약

step 1) Mapping: map의 결과물로 intermediate key value가 생성됨

step 2) Partitioning: 각각의 키 값을 기준으로 어느 리듀서에서 처리되어야 하는지 파티션을 결정함

step 3) Sorting: 같은 파티션 내의 값들은 키 값을 기준으로 sorting

step 4) Combining: 파티션 내의 값들에 대해 미리 reduce 작업을 진행하여 데이터의 양을 줄임

step 5) Spill to locak disk: 처리된 값들을 Mapper 노드의 디스크 공간에 저장

step 6) Notify Master: 매핑 과정을 완료하였다는 알림과 파티셔닝 정보를 마스터 노드에게 알림

step 7) Notify Reducers: 마스터 노드는 리듀서 노드들에게 이 정보를 전달

step 8) Remote Read: 리듀서 노드들은 자신에게 할당된 파티션들만 매퍼들에게서 읽어와서 리듀스 작업을 준비

STEP 3) Reduce, Write

다시 원래의 워드 카운팅의 사례로 돌아와서 셔플링 과정을 마친 데이터를 각각의 리듀서가 읽어와서 단어별 개수 카운팅을 완료하게 된다. 그리고 그 결과를 파일 형태로 출력하게 된다. 여기서 또 한 가지 문제가 있는데 리듀서들 역시도 분산된 노드들이라 리듀스 결과 데이터가 분산되어 있다는 것이다. 이를 다시 어떻게 취합하여 저장을 어떻게 하는지 의문이 생긴다. 아래 다이어그램은 리듀스 결과를 마친 데이터가 어떻게 저장되는 지에 대한 것을 도식화 한 것이다.

MapReduce는 reduce 결과로 생성된 Key Value 값들을 별도로 취합하지 않고 input file을 쪼개어 저장했을 때 처럼 분산 파일 시스템에 저장한다. 이는 reduce 결과물들을 다시 취합해야하는 수고를 덜어주고, 결과 데이터들을 가지고 다시 MapReduce 작업을 진행하기가 편리하다. 또한 읽기 작업을 수행할 때에는 분산 파일 시스템이 이 역할을 해주게 되어 MapReduce 자체의 구현이 더 간결해지는 장점이 있다.

• MapReduce 응용예시

앞에서 살펴본 워드 카운팅의 사례 외에도 MapReduce를 적용할 수 있는 사례들이 논문에 언급되어 있다. 앞서 언급했듯이 MapReduce 프레임워크에서 개발자는 Map과 Reduce 함수를 구현하며, 그 이외의 복잡한 분산 처리 및 저장을 MapReduce 프레임워크가 수행해주게 된다. 아래 사례들에서 K는 Key, V는 Value를 의미한다.

예시 1. Distributed Grep

1) Map: 주어진 텍스트 데이터 청크에서 정규표현식 패턴 매칭 수행, 패턴(K):1(V) 생성

2) Reduce: intermediate value를 복사하여 그대로 저장. 그리고 패턴 매칭만 수행하므로 별도의 sum 같은 로직을 수행하지 않음.

예시 2. Count URL Access Frequency

1) Map: 주어진 웹 페이지 요청 로그 데이터 청크에서 URL(K):1(V) 생성

2) Reduce: 찾아낸 K/V들을 모두 합쳐서 URL(K): total count(V) 생성

예시 3. Reverse Web-Link Graph

1) Map: 웹 페이지 html 데이터 청크에서 그 안에 포함된 URL을 찾아냄. Target(찾아낸 URL): Source(원본 URL) 생성

2) Reduce: 찾아낸 K/V를 합쳐서 Target: list(Source) 형태의 데이터 생성

예시 4. Term-Vector per Host

term vector란 문서 혹은 일련의 문서들에서 가장 중요한 단어들을 word(K): frequency(V) 형태로 계산한 것임

1) map: 문서 데이터 청크에서 각 단어별 빈도수를 계산하여 hostname(K): term vector(V)를 계산함.(term vector 자체도 K:V 데이터 이므로 K:(K:V) 형태가 되겠네요)

2) reduce: 앞서 구한 K/V 데이터를 모두 합쳐서 호스트 별로 term vector를 구합니다. hostname(K): term vector(V)

예시 5. Inverted Index

1) map: 웹 문서 데이터 청크를 입력 받아서 파싱한 다음 word(K): document ID(V) 형태의 데이터를 만들어줌.

2) reduce: 앞서 구한 K/V 데이터를 입력 받아서 다음과 같은 형태로 합쳐줌. word: list(document ID) 특정 단어가 쿼리로 들어오면 이렇게 구한 list(documentID)를 리턴해주면 간단한 검색 엔진을 구현할 수 있음.

예시 6. Distributed Sort

1) map: 레코드에서 소팅의 기준이 되는 키 값을 골라내어 key(K): record(V) 형태의 데이터를 만듬

2) reduce: 리듀스에서는 앞서 구한 K/V를 그대로 합쳐주어 파일에 저장합니다. 왜냐하면 Map 함수를 거쳐 생성한 intermediate file 들은 이미 소팅이 되어 있기 때문임.

• 맵리듀스 단점

단점 1. 복잡한 셔플링

앞서 살펴보았듯이 Map과 Reduce 과정은 상당히 단순하지만 이 둘 사이를 연결해주는 Shuffling 과정이 복잡하며 속도 또한 상당히 느린편임.

단점 2. 잦은 파일 입출력

또한 각각의 노드들은 map이나 reduce 작업을 하기 위해서 파일을 읽고 쓰는 작업을 해야함. Mapper는 분산 파일 시스템에서 데이터를 읽어와서 intermediate key value를 생성한 뒤, 자신의 디스크 공간에 파일을 씀. Reducer는 다시 이 파일을 읽어와서 리듀스 작업을 진행한 다음 결과 파일을 분산 파일 시스템에 출력함. 이렇게 반복적인 disk I/O는 성능 저하의 원인이 되며, 추후 spark가 이러한 문제점을 해결하기 위해 메모리 단에서 연산을 수행하는 방식을 제안하게 됨.

• HDFS 개요

HDFS는 대용량의 파일을 수천대의 컴퓨터를 묶어 구성한 클러스터에 안정적으로 저장해주는 storage 역할을 수행하며 하둡 에코 시스템에서 중요한 역할을 수행함.

• HDFS Architecture

HDFS는 클러스터에 파일을 분산하여 저장하고 이를 적절히 관리해주는 역할을 수행함.

전반적으로 HDFS는 Namenode, Datanode, Client 모듈로 구성되어 있음. Namenode는 전체적인 HDFS가 어떻게 구성되어 있는지, 어디에 어떤 블록이 저장되어 있는지 등 HDFS의 메타 데이터를 관리하는 역할을 수행함. Datanode는 실제적으로 파일을 저장하는 역할을 수행함. Client는 사용자가 작성한 프로그램으로 이 HDFS에 파일을 쓰거나 읽는 작업을 요청함.

• HDFS 파일 저장 방식

HDFS는 파일을 분산 저장하기 위해서 먼저 파일의 메타 데이터와 컨텐츠 데이터를 분리함. 메타 데이터는 파일의 접근 권한, 생성일, 수정일, 네임 스페이스 등 파일에 대해 설명하는 정보들이 해당됨. 컨텐츠 데이터는 실제 파일에 저장된 데이터임. 파일의 메타데이터는 네임 노드에 저장되며 컨텐츠 데이터는 블록 단위로 쪼개져서 데이터 노드에 저장됨. 블록의 기본 크기는 128MB이며 최소 3개의 복사본을 생성하여 분산 저장함. 아래 다이어 그램은 하나의 파일이 어떻게 블록 단위로 쪼개어지는 지 보여줌.

블록을 데이터 노드에 분산 저장하였기 때문에 네임 노드는 하나의 파일을 구성하는 블록들의 리스트와 어디에 저장되었는 지를 기록해주어야 함. 데이터 노드 역시 블록과 함께 블록에 대한 메타 데이터를 저장해줘야함.

• HDFS NameNode

네임 노드는 파일의 메타 데이터를 inode라는 곳에 저장함. inode란 unix 파일 시스템에서 사용되는 폴더 구조를 저장하는 구조체를 말함. 또한 네임 노드에는 파일 구성하는 블록들의 목록과 위치 정보가 저장되어 있음. 이러한 네임 노드는 HDFS에 파일을 읽거나 쓰는 작업의 시작점 역할을 수행함.

1) 파일 읽기 작업시 프로세스

step 1) 클라이언트는 네임 노드에 파일의 네임 스페이스에 해당하는 블록들의 목록과 주소를 요청

step 2) 클라이언트는 가장 가까이 위치한 데이터 노드에서 블록을 읽어옴

2) 파일 쓰기 작업

step 1) 클라이언트는 네임 노드에게 어드 데이터 노드에 블록을 쓰면 좋을지 요청

step 2) 네임 노드가 데이터 노드를 할당

step 3) 클라이언트는 블록을 쓰기 위한 데이터 파이프라인 생성

step 4) 데이터 파이프라인을 이용해서 블록 쓰기 작업 수행

• HDFS Datanode

파일의 컨텐츠 데이터는 블록 단위로 나뉘며, 하나의 블록은 적어도 3개의 복사본을 생성함. 데이터 노드는 그중 하나의 복사본을 저장하는 것임. 이러한 데이터 노드는 네임 노드와 3개지 종류의 통신을 하면서 커넥션을 유지함.

통신종류 1. handshake

맨 처음 데이터 노드를 네임 노드에 등록할 때 수행하는 통신임. handshake를 통해서 네임 노드는 데이터 노드가 통신이 가능하며 적절한 소프트웨어 버전인지 확인함. 적합한 데이터 노드임이 확인되면 데이터 노드를 식별할 수 있도록 고유한 id를 부여함.

통신종류 2. block report

네임 노드가 블록이 저장된 위치를 실제로 알 수 있는 것은 바로 이 block report 덕분임. 각 데이터 노드들은 매 시간마다 네임 노드에 블록이 실제로 저장되어 있는 위치와 현황을 네임 노드에게 알려줌. 네임 노드는 이 정보들을 취합하여 블록의 주소를 알게됨.

통신종류 3. heartbeat

하나의 HDFS 클러스터는 수천대의 컴퓨터로 구성되어 있으며, 이 중 일부 노드가 고장나는 상황은 빈번하게 발생함. 각 데이터 노드는 3초에 한 번씩 heartbeat를 네임 노드에게 보내어 자신이 잘 동작하고 있음을 확인시켜줌. 네임 노드는 10분간 응답이 없는 노드는 고장이 난 것으로 간주함. 그리고 해당 노드에 저장되어 있었던 블록들은 사용이 불가해지므로 다른 노드에 복사본을 하나씩 추가해줌.

• Name Node Backup & Availability

네임 노드의 데이터가 유실되거나 오염되면 HDFS 전체가 먹통이 될것이다. 이를 방지하기 위해서 Image, Checkpoint, Journal의 개념을 제시한다. Image란 파일의 메타 정보를 담은 inode와 각 파일을 구성하는 블록들의 리스트를 합쳐서 지칭하는 개념이다. 단 실제 블록이 저장된 위치 정보는 제외한다. 이는 계속해서 변화할 수 있기 때문이다. 이러한 image를 디스크에 쓴 것인 checkpoint이다. journal은 image에 발생한 모든 변화들을 기록한 로그 파일이다.

네임 노드는 가용성을 보장하기 위해서 주기적으로 checkpoint를 생성한다. 만일 네임 노드가 장애를 일으켜서 커졌다가 켜진 경우, 가장 최신 체크 포인트를 읽어온 다음 journal을 순차적으로 실행시켜서 복원한다. 체크 포인트 생성이 네임 노드에 부담이 될 수도 있으므로 별도의 백업 노드를 추가하여 네임 노드의 부담을 줄여주기도 한다.

그럼에도 네임 노드는 HDFS의 취약점으로도 지목된다. 한 대의 네임 노드가 너무 많은 역할을 수행하기 때문에 네임 노드가 장애를 일으키면 전체 클러스터가 먹통이 될 위험성이 있다는 것이다. 이를 방지하기 위해서 아래와 같은 구조를 설계하기도 한다.

먼저 Active Name Node를 배치하고, 이것이 장애를 일으길 켱우를 대비한 Standby Name Node를 마련한다. 그리고 앞서 살펴봤던 Backup Node의 역할을 수행하는 Journal Node를 NameNode에 연결해준다. 네임 노드는 이처럼 전체 HDFS를 조율하는 매우 중요한 역할을 담당함.

• Block Placement

블록들을 어떻게 배치할 것이냐도 파일의 availability, I/O performance 등을 결정하는 중요한 사안임. HDFS는 두 단계로 이루어진 추상화를 사용함.

각각의 데이터 노드들은 하나의 rack에 묶인다. rack이란 서버 컴퓨터들을 꽂아넣는 일종의 케이스를 의미하며 같은 렉에 꽂힌 데이터 노드들은 네트워크 스위치를 공유한다. 이 때문에 같은 랙에 속한 데이터 노드들 간에는 네트워크 통신이 빠르다. 여러 개의 랙들은 코어 스위치로 묶인다. HDFS는 이렇게 묶여있는 데이터 노드들에 아래와 같은 원칙으로 블록 replica들을 분산 시킨다.

원칙 1.하나의 데이터 노드에는 블록 replica를 하나만 저장한다.

원칙 2. 하나의 랙에는 블록 replica를 최대 두 개 까지만 저장한다.

이러한 원칙을 지켜서 블록을 분산시킬 경우 availability가 확보된다. 하나의 데이터 노드가 고장나거나 하나의 랙이 고장나더라도 여전히 히 다른 노드 혹은 다른 랙에 블록 replica가 저장되어 있기 때문이다. 그러므로 데이터의 유실을 방지할 수 있게 된다.

• Balancer

데이터 노드들을 운영하다보면 imbalance 문제가 발생한다. 나중에 추가된 데이터 노드의 디스크 공간은 비어있는 반면 오래 사용한 데이터 노드의 디스크 공간은 꽉 차게 되는 문제이다. 이를 방지하기 위해서 balancer 모듈이 있다. balancer는 데이터 노드들의 디스크 사용량이 비슷하게 데이터를 옮겨주는 역할을 한다. 그리고 그 와중에도 가용성을 떨어뜨리지 않도록 같은 노드에 두 replica를 저장하지 않고 같은 랙에 세 replica를 저장하지 않도록 최적화 시키는 역할을 수행한다.