[하둡에코시스템] 스파크 / apache spark

하둡의 문제점을 보완하기 위해 스파크 생김

 

하둡의 문제는

1. 반복적인 작업에는 비효율적임

2. 맵리듀스시 네트워크 트래픽으로 인해 성능저하됨.

 

스파크란?

기존 맵리듀스의 디스크 입출력을 보완하여

인 메모리기반 데이터 처리 프레임 워크 이다.

 

인 메모리 - 최초 데이터 입력, 출력에만 디스크에 작성함으로 네트워크 트래픽 발생 낮춤, 중간 결과는 병렬처리함

 

스파크의 주요기능 - 스파크 SQL, 스파크 스트리밍, 스파크 MLlib, 스파크 GraphX, 스파크 코어, 스파크 작업 처리

 

스파크 아키텍처

노드매니저 안에  드라이버 프로그램이 있음.

1. 드라이버 프로그램이 SparkContext 인스턴스 생성함(이때 yarn과 연결)

2. executors 를 요구

3. 애플리케이션 코드를 executors에 보낼 것

4. SparkContext 는 executors를 실행하기 위해 task를 보냄

 

Spark의 Driver는 YARN에서 Application Master와 같음

 

스파크 설치

$ wget https://downloads.apache.org/spark/spark-2.4.7/spark-2.4.7-bin-hadoop2.7.tgz
$ tar xvzf spark-2.4.7-bin-hadoop2.7.tgz
ln -s spark-2.4.7-bin-hadoop2.7/ spark

 

스파크의 RDD

RDD의 개념(Resilient Distributed Datasets)

-스파크 내에 저장되는 데이터 셋 타입

-내부적으로 연산하는 데이터들을 모두 RDD 타입으로 처리

Immutable, Partitioned Collections of Records

여러 분산 노드에 나누어지며

다수의 파티션으로 관리됨

변경이 불가능한 데이터 셋

 

RDD의 생성

1. 외부로부터 데이터를 가져올 때

2. 코드에서 생성되는 데이터를 저장할 때

 

RDD를 제어하는 2개의 연산 타입

1.Transformation : RDD에서 새로운 RDD 생성하는 함수 (filter, map)

2.Action : RDD에서 RDD가 아닌 다른 타입의 데이터로 변환하는 함수(count, collect)

 

RDD 분산 처리 방법

1.Immutable : 데이터셋 생성 뒤 변하지 않음

2.Partitoned : 데이터셋을 잘게 자름

 

RDD Partitioning

하나의 RDD는 여러 개의 파티션으로 나뉜다.

파티션의 개수와 파티션을 선택할 수 있다.

 

RDD Dependency

-Narrow Dependency

파티션이 1:1로 매핑 되어 네트워크 필요 없고 하나의 노드에서 작업 가능하다, 그리고 파티션 복원 쉬움  

-Wide Dependency

파티션이 1:N로 매핑 되어 파티션 재계산 비용 비싸며, 네트워크를 사용한다.

RDD Lineage

RDD연산 순서 기록 -> DAG (순환되지않음)

 Fault tolerant : 계보로 똑같은 RDD 생성 가능함

 Lazy execution

-Transformation 연산 시 계보 작정 됨

-Action 연산 시 계보 실행됨

미리 작성된 계보로 자원할당 참고할 수 있음

현재 쓰고있는 자원, 앞으로 사용할 자원, Dependency로 작업 스케줄링에 활용 가능함

 

Spark YARN 실행

scala : spark-shell --master yarn --queue queue_name
python : pyspark --master yarn --queue queue_name
 
--driver-memory 3G : spark driver가 사용할 메모리 default = 1024M
--executor-memory 3G : 각 spark executor가 사용할 메모리양
--executor-cores NUM : 각 spark executor의 코어의 양

Spark Shell 입력 후 코드 작성해보기

x=sc.parallelize([“spark”, ”rdd”, ”example”, “sample”, “example”], 3)
y=x.map(lambda x:(x,1))
y.collect()
[(‘spark’,1), (‘rdd’,1), (‘example’,1), (‘sample’,1), (‘example’,1)]

x에 스파크 콘텍스트 병렬화로 생성함

y에 x를 맵형식으로 x값과 1 저장함

collect 써서 집합 출력함

 

작성한 스파크에서 실행하는 방법

파이썬 파일  (num은 쓰레드 개수, default 값은 2~4개 정도)

spark-submit –master local[num] 파일명.py 

자바,스칼라 파일

spark-submit \ --class “SimpleApp”\ --master local[num] /location~/name.jar

 

스파크에서 맵리듀스

val input: RDD[(K1, V1)] = ...
val mapOutput: RDD[(K2, V2)] = input.flatMap(mapFn)
val shuffled: RDD[(K2, Iterable[V2])] = mapOutput.groupByKey().sortByKey()
val output: RDD[(K3, V3)] = shuffled.flatMap(reduceFn)

RDD (K1,V1) 데이터 입력

flatMap()연산 수행하여 RDD (K2,V2) 출력

RDD (K2,V2) 값으로 셔플함 , groupByKey()과 sortByKey()연산 수행함

RDD (K3, V3) 값에 shuffled후 flatMap() 연산한 결과값 저장함