[스파크] RDD 개요

RDD에 대해서는 지난 글에서도 간단히 살펴보았다.

https://thing-u.tistory.com/25?category=909947

[SPARK] SPARK DATAFRAME이란 - RDD와 어떻게 다른가?

 

이번 글에서는 RDD의 함수와 사용 방법을 알아보려한다.
 map, flatMap, filter, reduce등의 고차원 함수를 제공한다. 다음은 RDD클래스를 간단히 나타내본 코드이다. 

abstract class RDD[T]{
  def map[U](f:T=>U) :RDD[U]= ....
  def flatMap[U](f:T=>TraversableOnce[U]):RDD[U]= ...
  def filter(f:T =>Boolean) : RDD[T]=...
  def reduce(f:(T,T)=>T):T = ...
  def fold(f: T)(op:(T,T) =>T) : T 
  def aggregate[U](z:U)(seqop:(U,T) => U, combop:(U,U) =>U) : U 
}

 

RDD 생성방법

1. 기존에 존재하던 RDD 를 transforming
2. SparkContext or SparkSession object 에서 만들기

 - 함수 : parallelize : scala collection을 RDD로 변환
             textFile : HDFS나 filesystem 에서 읽어서 RDD[string]으로 반환 

[예시 맛보기]
1. 파일 READ
val licLines= sc.textFiles("/user/local/spark/LICENSE")
 -> LICENSE 파일 내용을 개행문자로 분할하여 만든 문자열의 컬렉션.
     licLines = RDD[String]

    
2. COUNT
val lineCnt = lincLines.count
 -> licLines 컬렉션에 포함된 줄 개수를 반환

3. FILTER
val bsdLines = licLines.filter(line => line.contains("BSD"))

 -> BSD 문자열이 등장한 줄만 포함하는 컬렉션
bsdLines.count
 -> 줄개수
3-1
* 기명함수를 사용하기
def isBSD(line: String) = {line.contains("BSD")}
val isBSD= (line:String)=>line.contains("BSD")

val bsdLines1 = licLines.filter(isBSD)

bsdLines1.count

 

4. 

bsdLine.foreach(bLine => println(bLine))
혹은 bsdLines.foreach(println)

 

RDD의 특징?

1) 불변성 : read only이다. 데이터를 조작할 때 새로운 RDD를 만들어 낼 뿐. 기존의 RDD가 변경되지않음.
2) 복원성 : 불변성을 가지기 때문에 이전 상태로 복원이 가능하다.
3) 분산 : 여러군데에 저장된 데이터들이 모여 하나의 RDD를 만든다.

복원성에 대해 조금 더 자세히 말해보자면,
스파크에서는 해당 데이터셋을 만드는데 사용된 변환 연산자 로그를 남긴다. 일부 노드에 장애가 발생한다면 이 로그를 이용해서 데이터셋을 다시 계산해 RDD를 복원한다.