[DEV] 14주차. Hadoop과 Spark (1)
1. 빅데이터의 정의와 예
빅데이터
- 서버 한 대로 처리할 수 없는 규모의 데이터
- Amazon의 data scientist 존 라우저가 내린 정의
- 분산 환경이 필요하느냐에 초점
- ex) pandas로 처리해야 할 데이터가 너무 커서 처리가 불가능하다면 ? -> spark!
- 기존의 소프트웨어로는 처리할 수 없는 규모의 데이터
- 대표적인 기존 소프트웨어: Oracle이나 MySQL과 같은 관계형 데이터베이스
- 분산 환경을 염두에 두지 않음
- Scale-up 접근 방식
- 메모리 추가, CPU 추가, 디스크 추가 등
빅데이터의 정의: 4V
- Volume : 데이터의 크기가 대용량인가
- Velocity : 데이터 처리 속도가 중요한가
- Variety : 데이터의 특성이 구조화/비구조화 혹은 둘 다
- Veracity : 데이터의 품질이 좋은가
예시 - 웹
- 웹 검색엔진 개발은 진정한 대용량 데이터 처리
- 웹 페이지를 크롤하여 중요한 페이지를 찾아내고 (페이지 랭크) 인덱싱하고 서빙
- 사용자 검색어와 클릭 정보 자체도 대용량
- 이를 마이닝하여 개인화 혹은 별도 서비스 개발 가능
- 검색어 기반 트렌드 파악, 통계 기반 번역, 등
- 이를 마이닝하여 개인화 혹은 별도 서비스 개발 가능
- 요즘은 웹 자체가 NLP 거대 모델 개발의 훈련 데이터로 사용되고 있음
2. 빅데이터 처리가 갖는 특징
어려움
- 큰 데이터를 손실없이 보관할 방법이 필요: 스토리지
- 처리 시간이 오래 걸림: 병렬 처리
- 이러한 데이터들은 비구조화된 데이터일 가능성이 높음: SQL만으로는 부족
- 예) 웹 로그 파일
해결 방안
- 큰 데이터 손실없이 보관
- 큰 데이터 저장이 가능한 분산 파일 시스템 필요
- 시간이 오래 걸림
- 병렬 처리가 가능한 분산 컴퓨팅 시스템 필요
- 비구조화 데이터
- 비구조화 데이터를 처리할 방법 필요
- 결국 다수의 컴퓨터로 구성된 프레임워크가 필요!
대용량 분산 시스템
- 분산 환경 기반 (1대 혹은 그 이상의 서버로 구성)
- 분산 파일 시스템과 분산 컴퓨팅 시스템이 필요
- Fault Tolerance
- 소수의 서버가 고장나도 동작해야 함
- 확장이 용이해야 함
- Scale-out이 가능하게
3. Hadoop
- 다수의 노드로 구성된 클러스터 시스템
- 마치 하나의 거대한 컴퓨터처럼 동작
- 사실은 다수의 컴퓨터들이 복잡한 소프트웨어로 통제됨
- Hadoop 1.0은 HDFS 위에 MapReduce라는 분산컴퓨팅 시스템이 도는 구조
- MapReduce 위에서 다양한 컴퓨팅 언어들이 만들어짐
- Hadoop 2.0에서 아키텍처가 크게 변경됨
- MapReduce의 생산성이 떨어짐
- 하둡은 YARN이라는 이름의 분산처리 시스템 위에서 동작하는 애플리케이션이 됨
- MapReduce보다 훨씬 일반적인 컴퓨팅 프레임워크
- MapReduce는 이 컴퓨팅 시스템 위에서 돌아가는 애플리케이션이 됨
- Spark은 YARN 위에서 애플리케이션 레이어로 실행됨
HDFS - 분산 파일 시스템
- 데이터를 블록 단위로 나누어 저장
- 블록 크기의 default값은 128MB
- 블록 복제 방식 (Replication)
- 각 블록은 3 곳에 중복 저장됨
- Fault Tolerance를 보장할 수 있는 방식으로 저장됨
- 하둡 2.0 네임노드 이중화 지원
- 네임노드: 다수의 Slave를 관리하는 Master
- 파일 정보, 파일을 구성하는 데이터 블록들이 어느 데이터 노드에 저장되어 있는지 등의 정보를 갖고 있는 디렉터리
- 네임노드가 동작하지 않으면 HDFS의 파일들은 아무 의미가 없음
- Active & Standby
- 둘 사이에 share edit log 존재
- Secondary Namenode는 여전히 존재
- 메인 네임노드의 정보를 주기적으로 복제
- 1.0에서는 메인 네임노드가 죽을 때 자동으로 secondary 네임노드가 실행되는 형태가 아니었음
- 2.0에서는 이중화 -> Active 네임노드에 문제가 생기면 Standby 네임노드가 자동으로 실제 네임노드 역할을 하도록 변경됨!
- 네임노드: 다수의 Slave를 관리하는 Master
MapReduce - 분산 컴퓨팅 시스템
- 하둡 1.0에서 처음 소개된 분산 컴퓨팅 시스템
- 하나의 잡 트래커와 다수의 태스크 트래커로 구성
- 잡 트래커: Master로, 일을 나눠서 다수의 태스크 트레커에게 분배
- 태스크 트래커에서 병렬 처리
- MapReduce만 지원
- 제너럴한 시스템이 아님
4. YARN의 동작 방식
분산 컴퓨팅 시스템: 하둡 2.0 (YARN 1.0)
- 세부 리소스 관리가 가능한 범용 컴퓨팅 프레임워크
- 리소스 매니저 : Master
- Job Scheduler, Application Manager
- 노드 매니저 : Slave
- 컨테이너 : 노드 매니저가 갖고 있는 자원 (Java의 JVM이라고 생각하면 됨)
- 앱 마스터
- 태스크
- 리소스 매니저 : Master
- Spark이 이 위에서 구현됨
YARN의 동작
1) 실행하고 싶은 코드와 환경 설정 정보 등을 RM에게 넘김
- 실행에 필요한 파일들은 application ID에 해당하는 HDFS 폴더에 미리 로딩이 되어있어야 함
2) RM은 지금 실행하고자 하는 애플리케이션의 Master를 만듦 : AM
3) 다수의 NM 중 비어있는 컨테이너가 있는 NM 하나를 임의로 골라서 컨테이너를 하나 달라고 요청
- AM은 프로그램마다 하나씩 할당되는 프로그램 마스터
4) 해당 컨테이너 안에 방금 클라이언트가 제출한 애플리케이션의 master 역할을 할 프로그램 실행
5) YARN Cluster 안에는 YARN application 수 만큼의 AM가 컨테이너 안에서 실행될 것
6) AM는 방금 클라이언트가 제출한 코드를 실행하기 위해 코드에 필요한 만큼의 자원을 RM에 요청 (컨테이너들: JVM (memory 등 JVM마다 할당되는 리소스들))
- RM은 data locality를 고려해서 리소스 (컨테이너)를 할당
7) AM은 NM로부터 할당받은 컨테이너 안에 실제 코드들을 돌리는 task 생성
- 이때 실행에 필요한 파일들이 HDFS에서 컨테이너가 있는 서버로 먼저 복사
8) task들은 주기적으로 본인의 상황을 AM에 보고 (heartbeat)
- 태스크가 실패하거나 보고가 오랜 시간 없으면 태스크를 다른 컨테이너로 재실행
- 기본적으로 데이터들을 HDFS에 있다고 가정
- YARN application (클라이언트): MapReduce, Spark 등
하둡 1.0 vs. 하둡 2.0
하둡 3.0
- YARN 2.0 사용
- YARN 프로그램들의 논리적인 그룹(플로우)로 나눠서 자원 관리가 가능
- 이를 통해 데이터 수집 프로세스와 데이터 서빙 프로세스를 나눠서 관리 가능
- 타임라인 서버에서 HBase를 기본 스토리지로 사용 (하둡 2.1부터)
- YARN 프로그램들의 논리적인 그룹(플로우)로 나눠서 자원 관리가 가능
- 파일 시스템
- 네임노드의 경우 다수의 standby 네임노드를 지원
- HDFS, S3, Azure Storage 외에도 Azure Data Lake Storage 등을 지원
5. MapReduce 프로그래밍
맵리듀스 프로그래밍 특징
- 데이터셋은 Key, Value의 집합이며 변경 불가 (immutable)
- 데이터 조작은 map과 reduce 두 개의 오퍼레이션으로만 가능
- 항상 하나의 쌍으로 연속으로 실행됨
- 이 두 오퍼레이션의 코드를 개발자가 채워야 함
- 맵리듀스 시스템이 Map의 결과를 Reduce 단으로 모아줌
- 같은 Key를 가진 값이 묶여서 전달
- 보통 map 코드가 실행되는 서버와 reduce 코드가 실행되는 서버가 다름
- 이 단계를 보통 셔플링이라고 부르며, 네트워크 단을 통한 데이터 이동이 생김
- 이 전송량에 따라 오퍼레이션 시간이 늘어날 수 있음
map, reduce
- Map :
(k, v) -> [(k', *v')*]- 입력은 시스템에 의해 주어지며, 지정된 HDFS 파일에서 넘어옴
- key, value 페어를 새로운 key, value 페어 리스트로 변환
- 출력은 동일한 Key, value 페어를 그대로 출력해도 되고, 출력이 없어도 됨
- Reduce :
(k', [v1', v2', v3', ...]) -> (k'', v'')- 입력은 시스템에 의해 주어짐
- 맵의 출력 중 같은 key를 갖는 key, value 페어를 시스템이 묶어서 입력으로 넣어줌
- key와 value 리스트를 새로운 key, value 페어로 변환
- SQL의
GROUP BY와 유사 - 출력이 HDFS에 저장됨
- 입력은 시스템에 의해 주어짐
shuffling and sorting
- Shuffling
- mapper의 출력을 reducer로 보내주는 프로세스
- 전송되는 데이터의 크기가 크면 네트워크 병목을 초래하고 시간이 오래 걸림
- Sorting
- 모든 mapper의 출력을 reducer가 받으면 이를 Key 별로 sorting
MapReduce: Data Skew
- 각 태스크가 처리하는 데이터 크기에 불균형이 존재한다면?
- 병렬처리가 큰 의미가 없음
- 가장 느린 태스크가 전체 처리 속도를 결정
- 특히 Reducer로 오는 나눠진 데이터의 크기는 큰 차이가 있을 수 있음
Group by나join등이 이에 해당- 처리 방식에 따라 reducer의 수에 따라 메모리 에러 등이 발생할 수 있음
- 데이터 엔지니어가 고생하는 이유 중 하나!
- sparkt, hive 등 빅데이터 시스템에는 이 문제가 모두 존재
MapReduce 프로그래밍의 문제점
- 낮은 생산성
- 2가지 오퍼레이션만 지원, 데이터 타입이 key-value 하나만 존재: 융통성 부족
- data skew가 발생하는 경우 튜닝/최적화가 쉽지 않음
- 배치 작업 중심
- 기본적으로 Low Latency가 아니라 Throughput에 초점이 맞춰짐
- 속도보다 크기에 집중
- 모든 입출력이 디스크를 통해 이루어짐
- Shuffling 이후에 Data Skew가 발생하기 쉬움
- Reduce 태스크 수를 개발자가 지정해주어야 함
MapReduce의 대안들
- 더 범용적인 대용량 데이터 처리 프레임워크들의 등장
- YARN, Spark
- SQL의 컴백: Hive, Presto 등의 등장
- Hive
- MapReduce 위에서 구현됨
- Throughput에 초점
- 대용량 ETL에 초점
- Presto
- Low Latency에 초점
- 메모리를 주로 사용
- Adhoc 쿼리에 적합
- AWS Athena가 Presto 기반
- Hive
6. Spark
Spark 3.0
- 다양한 분산환경 위에서 동작
- 가장 많이 쓰이는 것은 YARN
- 구성
- Spark Core
- Spark SQL
- Spark ML
- Spark Streaming
- Spark GraphX
Spark vs. MapReduce
- Spark는 기본적으로 메모리 기반 -> 데이터가 그렇게 크지 않을 때 빠른 속도에 유리
- 메모리가 부족해지면 디스크 사용
- MapReduce는 디스크 기반 -> 대용량 데이터에 유리
- MapReduce는 하둡(YARN) 위에서만 동작
- Spark는 하둡(YARN) 이외에도 다른 분산 컴퓨팅 환경 지원 (K8s, Mesos)
- MapReduce는 key, value 기반 데이터 구조만 지원
- Spark는 pandas 데이터프레임과 개념적으로 동일한 데이터 구조 지원
- Spark는 다양한 방식의 컴퓨팅 지원
- 배치 데이터 처리, 스트림 데이터 처리, SQL, ML, Graph 분석
Spark 프로그래밍 API
- RDD
- Resilient Distributed Dataset
- 로우레벨 프로그래밍 API로 세밀한 제어 가능
- 하지만, 코딩 복잡도 증가
- DataFrame & Dataset (pandas와 흡사)
- 하이레벨 프로그래밍 API로 점점 많이 사용되는 추세
- 구조화 데이터 조작이라면 보통 Spark SQL 사용 (join 등)
- 이 API가 꼭 필요한 경우
- ML Feature Engineering을 하거나, Spark ML을 사용하는 경우
- SQL만으로 할 수 없는 일의 경우
Spark SQL
- 구조화된 데이터를 SQL로 처리
- 데이터 프레임을 테이블처럼 SQL로 처리 가능
- 코드의 Readability 향상
- pandas도 동일 기능 제공
- 처음 나왔을 때는 Hive 쿼리보다 최대 100배까지 빠른 성능을 보장한다는 이야기가 있었지만,
- 사실은 그렇지 않음. Hive도 그 사이에 메모리를 쓰는 것으로 발전
- Hive: 디스크 -> 메모리
- Spark: 메모리 -> 디스크
- Presto: 메모리 -> 디스크
- 사실은 그렇지 않음. Hive도 그 사이에 메모리를 쓰는 것으로 발전
Spark ML
- 머신러닝 관련 다양한 알고리즘, 유틸리티로 구성된 라이브러리
- Classification, Regression, Clustering, Collaborative Filtering, ..
- 아직 오픈소스 spark ML에서 딥러닝은 지원하지 않는다고 생각하면 됨
- RDD 기반(spark.mllib)과 데이터프레임 기반(spark.ml)의 두 버전이 존재
- spark.mllib는 RDD 위에서 동작하는 이전 라이브러리로 더 이상 업데이트가 안됨
- 항상 spark.ml을 이용할 것 !
import pyspark.ml
- 장점
- 원스톱 ML 프레임워크!
- 머신러닝과 관계된 다양한 일들을 한 장소에서 할 수 있음
- 데이터프레임과 Spark SQL 등을 이용해 전처리 가능
- Spark ML을 이용해 모델 빌딩
- ML Pipeline을 통해 모델 빌딩 자동화
- MLflow로 모델 관리하고 서빙 (MLOps)
- 대용량 데이터도 처리 가능!
- 원스톱 ML 프레임워크!
Spark 데이터 시스템 사용 예시
- 기본적으로 대용량 데이터 배치 처리, 스트림 처리, 모델 빌딩
- 대용량 비구조화된 데이터 처리하기 (ETL / ELT)
- ML 모델에 사용되는 데이터 피쳐 처리 (배치 / 스트림)
- Spark ML을 이용한 대용량 훈련 데이터 모델 학습
- 대용량 비구조화된 데이터 처리 (Hive의 대체 기술)
- ETL / ELT
- 매우 큰 log 데이터가 S3에 적재
- S3: raw data가 쌓이는 데이터 레이크
- 주기적으로 데이터를 읽어서 spark로 정제 후 데이터 웨어하우스로 적재
- 혹은 다시 S3로
- ML 모델에 사용되는 대용량 피쳐 처리
- udemy 강의 추천
- 기록을 갖고 있는 사용자
- 과거 행동에서 피쳐 계산 & 현재 접속해서 무슨 행동을 했는지 보고 real-time 피쳐 계산
7. Spark 프로그램 실행 옵션
-
YARN 위에서 실행한다고 가정
- 개발/테스트/학습 환경 (Interactive Clients)
- 노트북 (주피터, 제플린)
- Spark Shell
- 프로덕션 환경 (Submit Job)
- spark-submit (command-line utility) : 가장 많이 사용됨
- 데이터브릭스 노트북
- 노트북 코드를 주기적으로 실행 가능
- REST API
- YARN이 아닌, Spark Standalone 모드에서만 가능 -> 사실 거의 없음
- API를 통해 Spark Job 실행
- 실행 코드는 미리 HDFS 등의 파일 시스템에 적재되어 있어야 함
Spark 프로그램 구조
- Driver
- 실행되는 코드의 마스터 역할 수행 (YARN의 AM)
- 사용자 코드를 실행하며 실행모드 (client, cluster)에 따라 실행되는 곳이 달라짐
- cluster 모드: driver가 YARN 클러스터 안에서 동작
- spark submit
- client 모드: driver가 YARN 클러스터 밖에서 동작
- 노트북, spark shell
- cluster 모드: driver가 YARN 클러스터 안에서 동작
- 코드르 실행하는데 필요한 리소스를 지정
--num-executors,--executor-cores,--executor-memory
- 보통 SparkContext를 만들어 Spark 클러스터와 통신
- 통신 대상
- Cluster Manager (YARN의 경우 Resource Manager)
- Executor (YARN의 경우 컨테이너)
- 통신 대상
- Executor
- 실제 태스크를 실행해주는 역할 수행 (YARN의 컨테이너)
- 하나의 Executor는 하나의 JVM이라고 보면 됨
- Transformations, Actions
Spark 클러스터 매니저 옵션
- local[n]
n: JVM 안에서 몇 개의 thread를 띄울 것인가 (몇 개의 CPU를 쓸 것인가)
- YARN
- cluster, client 모드
- Kubernetes
- Mesos
- Standalone (잘 쓰이지 않음)
local[n]
- 개발/테스트용
- Spark Shell, IDE, 노트북
n은 코어의 수- executor의 수가 됨
- local[]*
- 컴퓨터에 있는 모든 코어 사용
YARN
- 두 개의 실행모드가 존재
- Client 모드
- driver가 Spark 클러스터 밖에서 동작
- YARN 기반 Spark 클러스터를 바탕으로 개발/테스트/디버깅 등을 할 때 사용
- Cluster 모드
- driver가 Spark 클러스터 안에서 동작 (YARN - AM 안에서)
- 하나의 executor 슬롯을 차지
- 실제 프로덕션 운영에 사용되는 모드
- driver가 Spark 클러스터 안에서 동작 (YARN - AM 안에서)
요약
| 클러스터 매니저 | 실행모드 (deployed mode) | 프로그램 실행 방식 |
|---|---|---|
| local[n] | Client | Spark Shell, IDE, 노트북 |
| YARN | Client | Spark Shell, 노트북 |
| YARN | Cluster | spark-submit |