Spark RDD

Hadoop과는 다른 데이터 처리 방식을 채택한 Spark는 RDD(Resilient Distributed Dataset)라는 새로운 데이터 구조를 개발하였다. Spark에서 내부적으로 연산하는 데이터들은 모두 RDD타입으로 처리된다.

 

RDD의 특징

 

Immutable, Partitioned Collections of Record

 

• RDD 는 여러 분산 노드에 걸쳐서 저장되는 데이터(객체)의 집합이다.
• RDD는 데이터셋을 생성하여 메모리에 적재시키는 이후로는 변경이 불가능하다.
• RDD는 다수의 파티션으로 쪼개져 여러 분산 노드에 나뉘어져 관리된다.
• RDD는 데이터가 저장되어있는 File System에서 메모리상으로 데이터를 로딩할 때, 코드에서 생성되는 데이터를 메모리에 저장할 때 생성된다.

 

RDD 연산

RDD는 두 종류의 연산을 수행하며 단계적으로 데이터를 처리할 수 있다. Spark는 각 연산을 수행하는 수많은 함수들을 제공해준다.

Transformation: RDD에서 새로운 RDD를 생성하는 함수 타입이다.

• filter(): 기존 RDD에서 특정 데이터만 산출하여 RDD를 생성하는 함수이다.
• map(): 기존 RDD의 데이터를 분산하여 새로운 RDD에 배치하는 함수이다.

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

• count(): 변환 연산 후 파티션의 데이터 요소 개수를 반환하는 함수이다.
• collection(): 변환 연산 후 파티션의 데이터 요소 집합을 반환하는 함수이다.
•  

RDD 분산처리

 

RDD는 Partitioned, 데이터셋을 잘게 잘라 분산하여 가장 효율적으로 클러스터 노드에 분산시켜 저장한다.

 

큰 사이즈의 RDD는 여러개의 파티션으로 나누어져 여러 노드에 분산되어 처리될 수 있다. 파티션의 개수, 파티셔너의 종류(기본: Hash, Range / 사용자 정의)를 선택할 수 있다.

RDD의 변환 연산 이후 매핑되는 구조에 따라 dependency라는 두가지 형태로 구분된다. 하나의 파티션으로 매핑되는 Narrow dependency, 여러개의 파티션으로 매핑되는 Wide dependency가 있다.

Narrow dependency

• 변환 연산 이후 각 파티션은 N:1 또는 1:1으로 하나의 파티션에 매핑된다.
• map, filter, union 등의 작업으로, N:1 또는 1:1로 매핑되어 작업이 메모리 용량을 벗어나지 않아 하나의 노드에서 처리가 가능하다. 많은 자원을 사용하게 되는 클러스터 상 데이터의 셔플링이나 네트워크를 사용하지 않는다.
• 파티션에 문제가 생겨도 복원이 용이하다.

Wide dependency

• 변환 연산 이후 각 파티션은 1:N 또는 N:M으로 여러 파티션에 매핑된다.
• groupByKey, join 등의 작업으로, 메모리 용량을 벗어나 네트워크를 사용해 다른 노드와의 통신IO가 필요하며, 클러스터 상 데이터의 셔플링이 일어나는 작업들이 주된다.
• 파티션에 문제가 생기면 복원에 높은 비용이 든다.

Dependency는 작업의 최적화를 위해 분류되었으며, 많은 데이터를 Narrow dependency를 통해 처리한 후, Wide dependency를 적용함으로써 전략적인 네트워크 사용 및 용이한 복원을 얻을 수 있다.

 

RDD Lineage

 

RDD는 immutable, 변하지 않는 특성을 지닌다. 이는 재사용이 가능함을 뜻하고, RDD Lineage를 기억하는 개별 RDD를 통해 장애 발생 시 추적과 recovery에 용이해 Fault-tolerant함을 뜻한다.

 

RDD 연산은 DAG(Directed Acyclic Graph)구조를 띈다. 이 순서를 기록한것을 계보라는 뜻인 RDD Lineage라 부른다.

 

RDD Lineage 예시

계보를 기억해두면 장애로 인한 데이터 유실에도 동일한 RDD를 생성할 수 있어 Fault-tolerant를 확보한다. Wide dependency와 같은 높은 계산 비용을 가지는 RDD는 디스크에 Checkpoint를 남긴다.

Lazy Execution 특성을 지니는 RDD의 동작은, Transformation 연산을 읽어들일 땐 단순히 계보만 생성하며, 실제 Action 연산을 통해 데이터를 메모리에 로드하여 생성된 계보를 실행한다.

 

Reusing RDDs

Fault-tolerant 특징을 가지기 위해 RDD를 재사용하는데, Spark는 몇가지 옵션을 제공한다. 어느 옵션이든 공간적 혹은 연산적 비용을 지불하니, 상황에 맞게 전략적으로 재연산, 재사용을 취해야 한다.

 

Persist and Caching

RDD 영속화를 진행하면 실제 데이터를 구현하고 로드하여 재사용할 수 있게 구체화 하고, 유실에 대비해 Lineage 또한 저장한다.

StorageLevel이라는 속성을 인자로 받아 영속화 되는 저장장치 타입와 위치(메모리/디스크, 내부/외부), 직렬화, 복사본 개수 등을 설정할 수 있다.

더이상 쓰이지 않는 데이터가 자동으로 영속화 해제 되진 않기 때문에, unpersist함수를 호출하거나, executer 메모리 부족 시 LRU(Least Recently Used) Caching을 통해 순차적으로 영속 해제된다.

재연산을 방지하거나 긴 Lineage를 끊을 때 사용된다.

 

Checkpointing

RDD를 파일시스템에 직접 저장하는 방식으로 대용량의 영속화를 진행하여 재사용한다.

RDD가 존재하는 로컬 파일 시스템, 디스크에 직접 영속화할 수 있고, 떄로는 HDFS같은 분산 파일시스템에 영속화할 수 있다.

디스크IO를 통해 데이터를 처리하는 MapReduce의 단점을 그대로 가지게 되는 특징이 있지만, 작업의 중간결과를 저장하여 장애에 따른 고비용 재연산을 방지하는데 사용된다.

 

Shuffle files

Persist, Checkpoint의 직접적인 사용 외에도, Spark는 자체적으로 셔플 과정 중 디스크에 shuffle files라는 데이터를 기록하여 셔플 지점까지 재연산을 피할 수 있다.

이 데이터는 애플리케이션이 실행되는 동안 worker node의 로컬 디렉토리에 남아있게 된다.

중복작업으로 느린 작업에는 Persistence, 중간 실패가 우려되는 작업에는 Checkpointing을 고려할 수 있다.