[스파크] RDD, Dataframe, Dataset

(빅데이터 분석을 위한 스파크2 프로그래밍 1.1.5의 내용을 제 입맛대로 이해해본 내용입니다)

 

○모델

(프로그래밍에서) 현실 세계의 사물, 개념을 프로그래밍 언어로 설명하는 과정
데이터를 처리하기 위한 스파크 모델 -> RDD, Dataframe, Dataset
RDD는 스파크에서 데이터를 처리하기 위해 가장 처음으로 나온 모델
Dataframe과 Dataset은 RDD의 단점 혹은 아쉬운 점을 해소하기 위해 나온 모델

○RDD
"분산 데이터"에 대한 모델.
"값"으로 표현되는 데이터만 가리키는 것이 아님 -> 데이터를 다루는 방법까지 포함(Java의 클래스를 생각하면 된다)
"스파크가 사용하는 핵심 데이터 모델로서 다수의 서버에 걸쳐 분산 방식으로 저장된 데이터 요소들의 집합을 의미하며, 병렬처리가 가능하고 장애가 발생한 경우에도 스스로 복구될 수 있는 내성(Tolerance)를 가지고 있다"
RDD에 속한 요소들은 파티션으로 나뉘고 이를 스파크가 병렬로 나누어 처리함. 즉 여러 파티션들이 여러 노드들에서 나뉘어 처리되는 것. 
나뉘어 처리된 후에 결과들을 통합하는 과정에서 파티션에 속한 데이터들이 네트워크를 통해 다른 서버에 위치한 파티션으로 이동하는 과정(셔플링)이 발생할 수 있다.
셔플링은 전체적인 작업 성능에 큰 영향을 끼치기 때문에, 스파크에서는 몇몇 주요 연산에 파티션의 개수를 직접 지정할 수 있는 옵션을 제공함.
Resilient -> 회복력 있는. 복구 능력을 의미함.
스파크 RDD의 데이터 복구 과정은 RDD의 생성 과정에서 사용했던 작업 내용을 기억하고 있는 것. -> 문제가 발생하면 문제가 발생한 RDD를 생성했던 작업만 다시 수행해서 데이터를 복구함.
고로 RDD는 복구가 가능하기 위해서는 작업 과정에서 변경되면 안됨(읽기 전용). 변경이 가능하다면 모든 변경 이력을 저장하고 있어야 하기 때문.
예를 들어 txt 파일을 읽어 RDD를 생성한 후 추가 작업을 하게 되면 원본 RDD, 추가 작업 후의 RDD가 따로 생성되는 방식으로 작업이 수행됨.
RDD는 생성 과정을 기록하고 있고, 이는 Lineage라고 불림.

 

○RDD의 생성 방법 3가지

1) List, Set같은 메모리에 생성된 데이터 이용

메모리의 데이터를 RDD로 생성하는 방법

List, Set 등을 사용해 RDD를 생성한다.

 

2) File System이나 HDFS 데이터를 읽어 생성

로컬 파일시스템으로부터 RDD를 생성하는 방법

sc.textFile("경로")로 생성하면 된다.

"경로"에는 파일시스템, HDFS, S3, FTP 등을 사용할 수 있다.

Example)

val rddA = sc.textFile("file:///path/to/*");
 
val rddB = sc.textFile("hdfs://path/to/*");

 

* 주의할 점

HADOOP_HOME 환경 변수가 시스템에 등록되어 있는 경우 "file///", "hdfs://" 등의 스키마 정보 없이 "/path/to"의 경로만 지정할 경우 hdfs 시스템 경로로 인식됨. (물론 HADOOP_HOME 환경 변수 설정이 없다면 그림처럼 파일 경로로 인식함)

Window의 경우 Path에 역슬래시("\")가 사용되기 때문에 이를 java/scala에서 문자로 인식시키려면 역슬래시를 한번 더 붙여야 한다.

 

 

3) 기존 RDD로부터 수정해 새로운 RDD 생성
메모리로부터 RDD 생성 방법을 통해 만들어진 RDD를 변경했음.

기존의 RDD에 수정 작업이 생긴다면 나오게 되는 무조건적인 생성 방법.

RDD를 통한 RDD 생성

○RDD 생성 이후

RDD가 제공하는 다양한 연산 사용해 데이터 처리.

종류는 Transformation과 Action.

Transformation: RDD를 변형해 새로운 RDD를 생성하는 연산.
Action: Job의 변경 연산들을 종합해 처리하는 수행 연산.

 

* Transformation과 Action이 나뉘는 이유

스파크는 기본적으로 Lazy Evaluation 방식을 취하고 있기 때문이다.

Lazy Evaluation은 말 그대로 게으른 연산으로, Method 호출 과정 등 하나의 Method 내부에서 일어나는 모든 변환 과정을 개별적으로 실행하는 것이 아니라 "Action" 작업이 실행될 때 한꺼번에 실행하는 것을 의미한다.

위에서 연급한 Transformation이라는 것도 결국에는 변경 정보만 스파크에 전달하는 것이고 스파크에서는 변경 정보만을 가지고 있다가 Action이 호출되는 순간 변경 정보들을 수행하게 된다.

 

○스파크에서의 Lazy Evaluation 데이터 처리

Action이 호출되는 순간, 스파크는 변환 연산을 분석해 가장 최적의 변환 연산을 실행할 수 있음.

예를 들어 10개의 서버에 저장된 로그를 분석할 경우,

10개 서버의 데이터를 모두 모아 작업을 수행하는 것과 각 서버들에서 로그를 분석한 후 결과들을 합치는 것 중 

전자의 방법은 셔플이 대량으로 발생하게 되지만(대량의 데이터 이동에 따른),

후자의 경우 작업한 결과들을 합치기 때문에 전자에 비해 작업 효율이 훨씬 높다고 할 수 있음.

스파크는 대용량 처리를 위해 Action 단계에서 일종의 Query Plan을 짤 수 있는 Lazy Evaluation 방식을 채택함.

RDD 이후 Dataframe과 Dataset이 나타나게 된 이유는 스파크의 Query 최적화를 더 강화하기 위해서임.

 

○RDD가 제공하는 Method들

Action 연산은 연산의 결과가 RDD가 아닌 다른 값이거나, 아예 반환값이 없는 연산을 의미.

ex) RDD의 크기 계산->Long 타입의 결과값 반환, 결과 RDD를 타 저장소(혹은 스트림)이나 화면 출력.