[Spark] RDD 영속화(캐싱) - 이론 및 예제

RDD 영속화(캐싱) - 이론 및 예제 

동일한 RDD를 여러 번 사용하고 싶을 때도 있을 것이다. 생각없이 이를 시도한다면 스파크는 RDD와 RDD에서 호출하는 액션들에 대한 모든 의존성을 재연산하게 된다. 이는 데이터를 여러 번 스캔하는 반복 알고리즘들에 대해서는 매우 무거운 작업일 수 있다. 

RDD를 여러 번 반복 연산하는 것을 피하려면 스파크에 데이터 영속화(persist/persistence)를 요청을 할 수 있다. RDD 영속화에 대한 요청을 하면 RDD를 계산한 노드들은 그 파트션들을 저장하고 있게 된다. 영속화된 데이터를 갖고 있는 노드에 장애가 생기면 스파크는 필요 시 유실되ㄴ 데이터 파티션을 재연산한다. 만약 지연 없이 노드 장애에 대응하고 싶다면 데이터를 복제하는 정책을 선택할 수도 있다. 

스파크는 목적에 맞는 여러 수준의 영속화를 제공한다. 스칼라와 자바에서는 기본적으로 persist()가 데이터를 JVM 힙(heap)에 직렬화되지 않는 객체 형태로 저장한다. 파이썬에서는 영속화된 데이터는 늘 직렬화를 하므로 기본적으로 JVM 힙에 피클(파이썬 직렬화)된 객체가 저장된다. 데이터를 디스크나 오프힙(off-heap_ 저장 공간에 쓸 때는 데이터가 늘 직렬화된다.

MEMORY_ONLY는 공간 사용량 높고, cpu사용시간 낮고, 메모리에 저장하고, 디스크에 저장하지 않는다.

MEMORY_ONLY_SER 공간 사용이 낮고, cpu사용이 높고, 메모리에 저장하고, 디스크에 저장하지 않는다.

MEMORY_AND_DISK 공간 사용량이 높고, cpu사용이 중간, 메모리에 일부 저장, 디스크에 일부 저장 (메모리에 넣기에 데이터가 너무 많으면 디스크에 나눠 저장)

MEMORY_AND_DISK_SER 공간 사용량이 낮고, cpu사용이 높고, 메모리에 일부 저장, 디스크에 일부 저장 (메모리에 넣기에 데이터가 너무 많으면 디스크에 나눠 저장, 메모리에 직렬화된 형태로 저장)

DISK_ONLY 공간사용량이 낮고, cpu사용이 높고, 메모리에 저장하지 않고, 디스크에 저장

 

import org.apache.spark.storage.StorageLevel
val input = sc.parallelize(List(1, 2, 3, 4))
// input: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[35] at parallelize at <console>:31
val result = input.map(x => x*x)
// result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[36] at map at <console>:33
result.persist(StorageLevel.DISK_ONLY)
// res43: result.type = MapPartitionsRDD[36] at map at <console>:33

println(result.count()) // 4
println(result.collect().mkString(",")) // 1, 4, 9, 16
result.unpersist()
// res46: result.type = MapPartitionsRDD[36] at map at <console>:33

 

RDD의 persist()와 unpersist()를 통해 캐시에 데이터를 저장하고, 삭제를 한다. 파티션의 개수를 알고 싶으면 자바와 스칼라에서는 rdd.partitions.size, 파이썬에서는 rdd.getNumPartitions를 하면 된다.

 

[참고] 서적 - 러닝스파크

 

출처 : ourcstory.tistory.com/155?category=630696