[Spark] 빅데이터 처리와 Spark 소개(2)

 

1. 맵 리듀스 프로그래밍 소개
2. 스파크 소개 
3. 스파크 데이터 시스템 사용 예들
4. 스파크 실행 옵션
5. Summary

---

 

맵 리듀스 프로그래밍 소개

• 데이터 셋은 Key, Value의 집합이며 변경 불가(immutable)
• 데이터 조작은 map과 reduce 두 개의 오퍼레이션으로만 가능
  • 이 두 오퍼레이션은 항상 하나의 쌍으로 연속으로 실행됨
  • 이 두 오퍼레이션의 코드를 개발자가 채워야함
• 맵리듀스 시스템이 Map의 결과를 Reduce단으로 모아줌
  • 이 단계를 보통 셔플링이라 부르며 네트워크단을 통한 데이터 이동이 생김
  • 스파크도 비슷한 로직으로 구성된다.

맵리듀스 프로그래밍의 핵심: 맵과 리듀스

Map and Reduce

❖ Map: (k, v) -> [(k', v')*]

• 입력은 시스템에 의해 주어지며 입력으로 지정된 HDFS 파일에서 넘어옴
• 키,밸류 페어를 새로운 키,밸류 페어 리스트로 변환 (transformation)
• 출력:입력과 동일한 키,밸류 페어를 그대로 출력해도 되고 출력이 없어도 됨

❖ Reduce: (k’, [v1’, v2’, v3’, v4’, ...]) -> (k’’, v'')

• 입력은 시스템에 의해 주어짐
  • 맵의 출력중 같은 키를 갖는 키/밸류 페어를 시스템이 묶어서 입력으로 넣어줌
• 키와 밸류 리스트를 새로운 키, 밸류 페어로 변환
• SQL의 GROUP BY와 흡사
• 출력이 HDFS에 저장됨

• 데이터 치우침(skew)이 발생한다. 스파크에도 비슷한 현상이 생긴다.
• 최종파일 숫자는 리듀스 갯수 의 숫자가 동일하게 된다.

 

<Mapper>

1. 단어별로 Tokenizing을 실시하고 키,밸류 리스트를 만든다.
2. 단어별로 (key: 단어, val: 숫자) 의 형태가 페어로 들어가도록 구성한다.

<Reducer>

1. 같은 단어들에 대해서 리스트로 들어오게 된다.
2. 리스트에 서로 다른 값이 들어오게 된다면 모두 더해주는 작업을 해줌

아파치 하둡에서의 Map,Reduce 소스코드

 

MapReduce: Shuffling and Sorting

• Shuffling

  • Mapper의 출력을 Reducer로 보내주는 프로세스를 말함, 즉 매퍼의 아웃풋을 송신하는 과정이다.
    전송되는 데이터의 크기가 크면 네트워크 병목을 초래하고 시간이 오래 걸리게 된다.

• Sorting
  • 모든 Mapper의 출력을 Reducer가 받으면 이를 키별로 소팅
  • 하나의 맵과 리듀스로 출력을 원하는 결과를 얻기가 어렵다. 여러번의 반복을 통해 이뤄짐

 

MapReduce: Data Skew

• 매퍼 같은경우는 균등하게 분배되지만, 특정 키에따라 결과값이 치우치는 경우가 발생할 수 있다. 이를 Data Skew라고 함
• 각 태스크가 처리하는 데이터 크기에 불균형이 존재한다면?
  • 병렬처리의 큰 의미가 없음. 가장 느린 태스크가 전체 처리속도를 결정 (병목현상이 발생한다)
• 특히, Reducer로 오는 데이터 크기는 큰 차이가 있을 수 있음 (매퍼는 거의 발생하지 않음)
  • Group By나 Join등에 이에 해당함
  • 처리 방식에 따라 Reducer의 수에 따라 메모리 에러 등이 날 수 있음
  • 데이터 엔지니어가 고생하는 이유 중의 하나, 빅데이터 시스템에는 이 문제가 모두 존재 

 

MapReduce 프로그래밍의 문제점

1. 낮은 생산성
   • 프로그래밍 모델이 가진 융통성 부족 (2가지 오퍼레이션만 지원)
   • 튜닝/최적화가 쉽지 않음  예)데이터 분포가 균등하지 않은 경우
2. 배치작업 중심
   • 기본적으로 Low Latency가 아니라 Throughput에 초점이 맞춰짐 

MapReduce 대안들의 등장

 

1. 더 범용적인 대용량 데이터 처리 프레임웍들의 등장 YARN, Spark
2. SQL의 컴백: 구조화된 데이터를 처리하는데 있어서는 최고.. Hive, Presto등이 등장!
   • Hive
     • MapReduce위에서 구현됨.
     • Throughput에 초점. 대용량 ETL에 적합
   • Presto
     • Low latency에서 초점. 메모리를 주로 사용.
     • Ad-hoc 쿼리에 적합함, AWS Athena가 Presto 기반

 

Hive 와 Presto의 차이점?

• Hive는 하둡1.0에서는 MR위에 있었지만, 2.0에서 'Tez' 위에서 구현되어 ETL에 성능의 초첨을 맞춤
• 과거에 Presto는 Ad-hoc query를 통해 스피드에 초점을 맞춘 경향이 더 강했다.
• 최근에는 Hive도 Ad-hoc query를 지원하고, Presto도 ETL을 지원하기에 서로 비슷한 기능을 지원하게 되었다.

 

스파크 소개 / 실행 옵션

• 버클리 대학의 AMPLab에서 아파치 오픈소스 프로젝트로 2013년 시작
• 나중에 Databricks라는 스타트업 창업 하둡의 뒤를 잇는 2세대 빅데이터 기술

• YARN등을 분산환경으로 사용함, Scala로 작성됨
• 빅데이터 처리 관련 다양한 기능 제공을 하나의 패키지에서 제공된다.
  

 

Spark 3.0의 구성

Standalone Scheduler은 개발용 이외에는 쓰이지 않음, Mesos K8s 등의 컨테이너 위에서 스파크가 구성된다.

• Spark Core
• Spark SQL
• Spark ML(DataFrame 기반), MLlib (RDD기반 - 요즘은 없어지는 단계)
• Spark Streaming
• Spark GraphX

Spark vs. MapReduce

• Spark은 기본적으로 메모리 기반
  • 메모리가 부족해지면 디스크 사용
  • MapReduce는 디스크 기반 메모리에 보존하는게 아니라 디스크에 저장한다. (큰 데이터 처리에는 좋은데 속도가 느리다.)
• MapReduce는 하둡(YARN)위에서만 동작, MapReduce는 키, 밸류 기반 데이터 구조만 지원하기 때문에 배치처리만 가능하다.
• Spark은 하둡(YARN)이외에도 다른 분산 컴퓨팅 환경 지원 (K8s, Mesos)
• Spark은 판다스 데이터프레임과 개념적으로 동일한 데이터 구조 지원
• Spark은 다양한 방식의 컴퓨팅을 지원한다: 배치 데이터 처리, 스트림 데이터 처리, SQL, 머신 러닝, 그래프 분석

Spark프로그래밍API

• RDD (Resilient Distributed Dataset) : 이거로 프로그래밍 하는 사람 많지않음 
  • 로우레벨 프로그래밍 API로 세밀한 제어가 가능, 하지만 코딩 복잡도 증가

• DataFrame & Dataset (판다스의 데이터프레임과 흡사) : 실제 프로그래밍에서 자주 쓰임 (DF: Python, DS: Java, Scala ) 
  • 하이레벨 프로그래밍 API로 점점 많이 사용되는 추세
  • 구조화 데이터 조작이라면 보통 Spark SQL을 사용(주로, data aggregation, data set 등의 Join 등을 할 때..)
• DataFrame/Dataset이 꼭 필요한 경우는?
  • ML 피쳐 엔지니어링을 하거나 Spark ML을 쓰는 경우
  • SQL만 으로 할 수 없는 일의 경우

Spark SQL

• Spark SQL은 구조화된 데이터 처리를 SQL로 처리
• 데이터 프레임을 SQL로 처리 가능
  • 데이터프레임은 테이블처럼 SQL로 처리 가능
  • 판다스도 동일 기능 제공
• Hive 쿼리 보다 최대 100배까지 빠른 성능을 보장?
  • 더이상 그렇지 않다. Hive도 그 사이에 메모리를 쓰는 걸로 발전('Tez'위에서 돌아가는데 메모리와 디스크 동시에 사용한다.) 
    • Hive: 디스크 -> 메모리
    • Spark: 메모리 -> 디스크
    • Presto: 메모리 -> 디스크 
  • 시작점은 서로 달랐지만, 다 고만고만해졌다.

Spark ML

• 머신러닝 관련 다양한 알고리즘, 유틸리티로 구성된 라이브러리
  • Classification, Regression, Clustering, Collaborative Filtering, ...
• 전체 리스트는 링크 참고. 딥러닝 지원은 미약
• RDD 기반과 데이터프레임 기반의 두 버전이 존재
  • spark.mllib vs. spark.ml
  • spark.mllib가 RDD 기반이고 spark.ml은 데이터프레임 기반
  • spark.mllib는 RDD위에서 동작하는 이전 라이브러리로 더 이상 업데이트가 안됨
• 항상 spark.ml을 사용할 것! : import pyspark.ml (import pyspark.mllib)

Spark ML의 장점

• 원스톱 ML 프레임웍! 대용량 데이터도 처리 가능!
  • 데이터프레임과 SparkSQL등을 이용해 전처리
  • Spark ML을 이용해 모델 빌딩
  • ML Pipeline을 통해 모델 빌딩 자동화
  • MLflow로 모델 관리하고 서빙 (MLOps)
• 서버 한개로 돌아가는 'scikit-learn' 과 비교했을 때 여러서버를 돌릴 수 있는것은 큰 차이점이라고 볼 수 있다.

 

Spark 데이터 시스템 사용 예들

• 기본적으로 대용량 데이터 배치 처리, 스트림 처리, 모델 빌딩
  • 예1) 대용량 비구조화된 데이터 처리하기 (ETL 혹은 ELT)
  • 예2) ML 모델에 사용되는 대용량 피쳐 처리 (배치/스트림)
  • 예3) Spark ML을 이용한 대용량 훈련 데이터 모델 학습

 

Spark 데이터 시스템 사용 예1

• 대용량 비구조화된 데이터 처리하기 (Hive의 대체 기술)
• ETL 혹은 ELT

 

 

 

Spark 데이터 시스템 사용 예2

• ML 모델에 사용되는 대용량 피쳐 처리
• 배치형태(Redshift, S3)로 미리계산, 또는 실시간계산(Kafka, Kinesis)
• 피쳐계산을 별도의 시스템 뒤에서 진행된다.
• 계산된 피쳐를 NoSQL에다가 저장해둔다. (데이터 사이즈가 작으면 Redis 사용도 가능)

 

 

 

스파크 실행 옵션

Spark 프로그램 실행 환경

• 개발/테스트/학습 환경 (Interactive Clients: 명령하나 실행하고 결과보고)
  • 노트북 (주피터, 제플린)
  • Spark Shell
•  프로덕션 환경 (Submit Job: 주기적으로 실행, 필요할 때 실행)
  • spark-submit (command-line utility): 일반적으로 가장 많이 사용됨
  • 데이터브릭스 노트북: 노트북 코드를 주기적으로 실행해주는 것이 가능
  • REST API: 
    • (Yarn위에서가 아닌)Spark Standalone 모드에서만 가능(프로덕션에서는 사용되지 않는다)
    • API를 통해 Spark 잡을 실행
    • 실행코드는 미리 HDFS등의 파일 시스템에 적재되어 있어야함

Spark 프로그램의 구조

 

• Driver: 실행되는 코드의 마스터 역할 수행 (YARN의 Application Master)
  • 사용자 코드를 실행하며 실행 모드(client, cluster)에 따라 실행되는 곳이 달라짐
    • client 모드 : Yarn 클러스터 밖에서 실행한다. (노트북, Spark-shell)
    • cluster 모드 : Yarn 클러스터에서 안에서 컨테이너를 하나 잡아다가 실행한다.
  • 코드를 실행하는데 필요한 리소스를 지정함
    
    • --num-executors : 몇개의 executor 쓸건지 수를 정해주기
    • --executor-cores : executor 마다 cpu몇개 쓸건지
    • --executor-memory : executor 마다 memory 몇개 쓸건지
  • SparkSession을 만들어 Spark 클러스터와 통신 수행
    • Cluster Manager (YARN의 경우 Resource Manager)
    • Executor (YARN의 경우 Container)
  • 사용자 코드를 실제 Spark 태스크로 변환해 Spark 클러스터에서 실행

 

• Executor: 실제 태스크를 실행해주는 역할 수행 (YARN의 컨테이너 Slave)
  • 실제 태스크를 실행해주는 역할 수행 (JVM): Transformations, Actions
  • YARN에서는 Container가 됨

 

 

Spark 클러스터 매니저 옵션

1. local[n] : JVM하나짜리 클러스터 - 실습환경 ([n] : 몇개의 스레드를 쓸건지(스파크 클러스터마냥))
2. YARN : 클라이언트 모드, 클러스터 모드
3. Kubernetes
4. Mesos
5. Standalone : 많이쓰이지 않음

 

• local[n]: 개발/테스트용 (Spark Shell, IDE, 노트북)
  • 하나의 JVM이 클러스터로 동작: Driver와 하나의 Executor 실행
  • n은코어의수 : Executor의 스레드 수가 됨
  • local[*]는 무엇일까? : 컴퓨터에 있는 모든 코어 사용 (cpu 숫자 만큼 스레드 만들어라)

 

• YARN
  • 두 개의 실행 모드가 존재: Client vs. Cluster
  • Client 모드: Driver가 Spark 클러스터 밖에서 동작
    • YARN 기반 Spark 클러스터를 바탕으로 개발/테스트 등을 할 때 사용
  • Cluster 모드: Driver가 Spark 클러스터 안에서 동작
    • 하나의 Container 슬롯을 차지함, 실제 프로덕션 운영에 사용되는 모드
    • 실제 프로덕션 운영에 사용되는 모드

 

◆ Spark 클러스터 매니저와 실행 모델 요약

 

클러스터 매니저	실행 모드 (deployed mode)	프로그램 실행 방식
local[n]	Client	Spark Shell, IDE, 노트북
YARN	Client	Spark Shell, 노트북
YARN	Cluster	spark-submit

 

 

---

요약...

• 빅데이터의 처리를 위해 하둡이라는 오픈소스가 등장
• 분산 파일 시스템과 분산 컴퓨팅 시스템으로 구성: HDFS와 맵리듀스/YARN
• 맵리듀스 프로그래밍의 제약성으로 인해 SQL이 재등장
• Spark은 대세 대용량 데이터 분산 컴퓨팅 기술
  • Pandas + Scikit Learn의 스테로이드 버전
  • SQL과 스트림 데이터와 그래프 처리도 제공