Spark - RDDs
Resilient Distributed Datasets 즉, RDD는 변경할 수 없는(immutable) elements 집합이다.
간단히 RDD의 특징을 살펴보면
- Partitioning
- Persistence
- Fault Tolerance
- Lazy Evalution
- ...
이 밖에도 많이 있지만, 자세한 것은 처음 접할 때 어려울 수 있으니 차근차근 알아보도록 하겠다.
다만, 이와 같은 특징들은 Spark가 데이터를 효과적으로 처리할 수 있도록 한다.
*RDD의 원천부터 알고 싶다면, "https://www.usenix.org/system/files/conference/nsdi12/nsdi12-final138.pdf" 관련 Paper를 읽어보는 것을 추천한다.
위에서 언급했듯이, RDD는 여러가지 데이터 타입들의 집합이다. 코드를 잠깐 들춰보면,
abstract classRDD[T] extends Serializable with Logging
abstract class의 형태로 구성되어 있으며, 또한 General한 타입으로 되어있어서 유연하게 데이터를 관리할 수 있게되어 있다. (다양한 타입의 객체 저장)
* RDD의 문서는 "https://spark.apache.org/docs/0.7.3/api/core/spark/RDD.html"
정의되어 있는 RDD를 살펴보면 map, filtermap, collect, ...등등의 다양한 method들을 정의하고 있다.
그런데 이런 method들은 크게 transformation과 action 두가지로 분류할 수 있다.
- Transformation : 결과를 RDD로 반환하며, 즉시 computed 되지 않는다. (Lazy)
- Action : RDD에 기반하여 결과를 즉시 반환하거나 외부 파일시스템에 저장한다. (eager)
Transformation |
Action |
- map - flapmap - filter - distinct - union - intersection - ... |
- collect - count - take - reduce - foreach ... |
그러나 RDD의 모든 작업은 action이 수행 될 때마다 계산된다. 이는 시간 및 비용이 드는 작업이다.
예를 들어, 특정 데이터셋을 바탕으로 작업을 할 때 사전 작업이 동일하다면 action 작업을 수행할 때마다 동일한 작업을 매번 하게 된다.
이런 불필요한 작업을 방지할 수 있도록, RDD는 persist(), cache() method를 제공한다.
val lastYearsLogs: RDD[String] = ... val logsWithErrors = lastYearsLogs.filter(_.contains("ERROR")).persist() val firstLogsWithErrors = logsWithErrors.take(10) |
위 코드를 살펴보면, logsWithErrors는 메모리에 값이 저장되어 있기 때문에 이후 작업은 해당 메모리로 부터 시작하면 된다. (처음부터 다 할 필요가 없다.)
덧붙여, cache는 Spark 내부 Storage Level의 디폴트 값에 근거하여 저장하는 방식이다.
(persist와의 차이라면 persist는 Storage Level를 사용자 임의로 설정할 수 있다. *아래 참조)
I. MEMORY_ONLY: In this level, RDD object is stored as a de-serialized Java object in JVM. If an RDD doesn’t fit in the memory, it will be recomputed. II. MEMORY_AND_DISK: In this level, RDD object is stored as a de-serialized Java object in JVM. If an RDD doesn’t fit in the memory, it will be stored on the Disk. |
Reference
- https://www.quora.com/What-are-different-Persistence-levels-in-Apache-Spark