[논문 리뷰] Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing

2022-08-14 13 분 소요

“Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing” 논문¹을 정리 및 해설한 글이다.

대규모 분산 컴퓨팅 환경에서 반복 연산(Iterative Machine Learning)과 대화형 쿼리(Interactive Data Mining)의 병목은 데이터 재사용 시 발생하는 디스크 I/O와 직렬화 오버헤드였다. 본 논문은 연산 계보(Lineage)를 기록해 장애 시 손실된 파티션만 결정적으로 복구하는 불변 분산 메모리 추상화인 RDD(Resilient Distributed Datasets)를 제안함으로써, 메모리 복제 비용을 들이지 않고 장애 복구 가능한 인메모리 분산 연산을 실현했다.

Contents

1. Introduction
2. Resilient Distributed Datasets
3. Spark Programming Interface
1. Representing RDDs
1. Implementation
1. Evaluation
1. Discussion (한계 및 트레이드오프)

논문 출처 : NSDI 12 paper - RDDs

1. Introduction

대규모 분산 환경에서 반복 알고리즘과 대화형 쿼리를 가속하기 위해서는 효율적인 클러스터 인메모리 추상화가 필수적이다.

맵리듀스(MapReduce)같은 클러스터 컴퓨팅 프레임워크는 대규모 데이터 분석에 널리 채택되었다.
이런 시스템을 통해 사용자는 작업 분산(work distribution)과 장애 허용(fault tolerance)에 대해 걱정할 필요 없이 높은 수준의 연산자 집합을 사용하여 병렬(parallel) 연산을 작성할 수 있다.
하지만, 현재 프레임워크는 클러스터의 계산 리소스에 액세스하기 위한 수많은 추상화(abstraction)를 제공하지만 분산 메모리(distributed memory)를 활용하기 위한 추상화는 부족하다.
- 추상화(abstraction) : 복잡한 자료, 모듈, 시스템 등으로부터 핵심적인 개념 또는 기능을 간추려 내는 것을 말한다. ²
이로 인해 여러 계산에서 중간 결과(intermediate result)를 재사용(reuse)하는 중요한 클래스의 애플리케이션에는 비효율적이다.
- 데이터 재사용은 PageRank, K-means 클러스터링, 로지스틱 회귀 분석을 포함한 많은 반복 기계 학습 및 그래프 알고리즘에서 일반적이다.
- 사용자가 데이터의 동일한 하위 집합에 대해 여러 Adhoc 쿼리(query)를 실행하는 대화형 데이터 마이닝에도 그렇다.
  - Adhoc query : 그때그때 만들어 쓰는 쿼리. 즉, 서버에 캐시되지 않고 즉석에서 만들어진 쿼리 ³

기존 시스템은 잡(Job) 간 데이터 공유를 분산 파일 시스템에만 의존했기 때문에 극심한 I/O 병목이 발생했다.

하지만, 대부분의 프레임워크에서 계산 간(ex. 맵리듀스 작업 간) 데이터를 재사용할 수 있는 유일한 방법은 외부 안정적인 스토리지 시스템(ex. 분산 파일 시스템)에 데이터를 쓰는 것이다.
이로 인해 데이터 복제, 디스크 I/O 및 직렬화(serialization)로 인해 추가적인 I/O 오버헤드가 발생하므로 애플리케이션 실행 시간이 길어질 수 있다.
- 직렬화(serialization) : 데이터 구조나 오브젝트 상태를 동일하거나 다른 컴퓨터 환경에 저장(ex. 파일이나 메모리 버퍼에서 또는 네트워크 연결 링크 간 전송)하고 나중에 재구성할 수 있는 포맷으로 변환하는 과정 ⁴
  - 오브젝트를 직렬화하는 과정은 오브젝트를 마샬링(marshalling)이라 한다.
따라서 저자들은 데이터 재사용이 필요한 애플리케이션을 위한 전문 프레임워크를 개발했다.
본 논문에서는 광범위한 애플리케이션에서 효율적인 데이터 재사용을 가능하게 하는 탄력적 분산 데이터 세트(Resilient Distributed Datasets, RDDs)라는 새로운 추상화를 제안한다.
- Resilient : 메모리 내부에서 데이터 손실 시 유실된 파티션을 재연산해 복구할 수 있다.
- Distributed : 스파크 클러스터를 통하여 메모리에 분산되어 저장된다.
- Data : 파일, 정보 등등
RDDs는 사용자가 메모리에 중간 결과(intermediate result)를 명시적(explicitly) 유지하고(persist), 데이터 배치(data placement)를 제어하기 위해 파티셔닝(partitioning)을 설정하고 풍부한 연산자 집합을 사용하여 조작할 수 있도록 하는 장애 허용(fault tolerance) 병렬(parallel) 데이터 구조이다.
- 파티셔닝(partitioning) : 데이터베이스를 여러 부분으로 분할하는 것이다. 데이터가 너무 커졌을 때, 조회하는 시간이 길어졌을 때 또는 관리 용이성, 성능, 가용성 등의 향상을 이유로 행해진다. ⁵

RDD 개념 다이어그램 이미지 출처 ⁶

RDDs를 설계할 때의 주요 과제는 fault tolerance를 효율적으로 제공할 수 있는 프로그래밍 인터페이스를 정의하는 것이다.
분산 공유 메모리(distributed shared memory), 키/값 저장(key-value store), 데이터베이스(database) 및 Piccolo와 같은 클러스터의 인메모리(in-memory) 스토리지를 위한 기존 추상화는 가변(mutable) 상태에 대한 세분화된 업데이트를 기반으로 인터페이스를 제공한다.

분산 메모리 기존 추상화 예시 이미지 출처 ⁷

이 인터페이스를 사용할 경우 fault tolerance를 제공하는 유일한 방법은 시스템 간에 데이터를 복제하거나 시스템 간에 업데이트를 기록(log)하는 것이다.
- 비용이 너무 많이 든다.
  
  이미지출처 ⁸

데이터 복제 대신 변환 계보를 기록하는 역발상이 RDD의 핵심이다.

RDDs는 많은 데이터 항목에 동일한 작업을 적용하는 coarse-grained 변환(transformation) (ex. map, filter, join)을 기반으로 인터페이스를 제공한다.
이를 통해 실제 데이터가 아닌 데이터 세트(lineage)를 구축하는 데 사용되는 변환(transformation)을 로깅(logging)하여 장애 허용을 효율적으로 제공할 수 있다.
- 만약 RDDs의 파티션이 손실된다면, RDDs는 그것이 어떻게 다른 RDDs로부터 파생되었는지에 대한 충분한 정보를 가지고 그 파티션만을 재계산(recompute)한다.
- 따라서 손실된 데이터를 복제 비용이 많이 들지 않고 신속하게 복구할 수 있다.
RDDs는 많은 병렬 애플리케이션에 적합한데, 이러한 애플리케이션들은 자연스럽게 여러 데이터 항목에 동일한 작업을 적용하기 때문이다.

이제 이러한 RDD 추상화가 실제로 어떤 속성을 가지며 어떻게 정의되는지 자세히 살펴본다.

2. Resilient Distributed Datasets (RDDs)

RDD는 결정적 변환으로 생성되는 불변 파티션 레코드의 모음으로, 계보를 통해 무복제 결함 감내를 달성한다.

2.1 RDD Abstraction

RDD 추상화 구성 요소 이미지출처 ⁹

RDD는 읽기 전용(read-only), 분할된 레코드(record)의 모음이다.
- 레코드(record) : 데이터베이스에서 하나의 단위로 취급되는 자료의 집합 (행과 같음) ¹⁰
RDD는 1) 안정적인 스토리지 안의 데이터 또는 2) 다른 RDDs에 대한 결정적 연산(deterministic operation)을 통해서만 생성될 수 있다.
- 결정적(deterministic) : 예측한 그대로 동작. 어떤 특정한 입력이 들어오면 언제나 똑같은 과정을 거쳐서 언제나 똑같은 결과를 내놓는다. ¹¹
이러한 작업을 RDD의 다른 작업과 구별하기 위해 변환(transformation)이라 부른다.
- 변환의 예로는 map, filter, join 등이 있다.
RDD는 다른 데이터 세트(ex. 계보(lineage))에서 어떻게 파생되었는지에 대한 충분한 정보를 가지고 있어 안정적인 저장소의 데이터로부터 파티션을 계산한다.
사용자는 재사용할 RDD를 지정하고 이를 위한 스토리지 전략(ex. in-memory 스토리지)을 선택할 수 있다.
또한 RDD의 요소들이 각 레코드의 키(key)에 기초하여 기계들 사이에 분할되도록 요청할 수 있다.

2.2 Spark Programming Interface

스파크 프로그래밍 모델은 RDD 정의를 위한 변환과 실제 연산을 촉발하는 액션으로 나뉜다.

프로그래머는 안정적인 스토리지의 데이터에 대한 변환(ex. map, filter)을 통해 하나 이상의 RDD를 정의하는 것으로 시작한다.
그런 다음 이러한 RDD를 애플리케이션에 값을 반환하거나 데이터를 스토리지 시스템으로 내보내는 연산(operation)에 사용할 수 있다.
- count : 데이터셋의 요소 수를 반환
- collect : 요소 자체를 반환
- save : 데이터셋을 스토리지 시스템으로 출력
spark는 기본적으로 메모리에 지속적인(persistent) RDD를 유지하지만 RAM이 충분하지 않으면 디스크에 지속적인 RDD를 보낼 수 있다.

2.2.1 Example: Console Log Mining

대용량 로그 파일에서 오류 메시지만을 메모리에 로드하여 빠른 대화형 검색을 수행할 수 있다.

웹 서비스에서 오류가 발생하고 있으며 운영자가 원인을 찾기 위해 HDFS에서 테라바이트의 로그를 검색하려고 한다고 가정한다.

스파크를 사용하면 운영자는 로그의 오류 메시지만 노드(node) 집합에 걸쳐 RAM으로 로드하고 대화식으로(interactively) 쿼리(query)할 수 있다.

lines = spark.textFile("hdfs:/...") // HDFS 파일로 백업된 RDD를 정의
errors = lines.filter(_.startsWith("ERROR")) // 필터링된 RDD를 파생
errors.persist() // 오류가 메모리에 남아 쿼리 간에 공유될 수 있도록 요청

현재 클러스터에 대해 수행될 작업이 없다면 RDD를 사용하여 메시지 수를 셀 수 있다.
```
errors.count()
```

사용자는 다음과 같이 RDD에서 추가 변환을 수행하고 결과를 사용할 수도 있다.

errors.filter(_.contains("MySQL")).count()  // MySQL에서 언급한 에러 세기

// HDFS를 배열(array)로 언급하는 오류의 time field를 반환한다.
// 시간이 탭으로 구분된 형식의 필드 번호를 3이라 가정
errors.filter(_.contains("HDFS"))
      .map(_.split('\t')(3))
      .collect()

오류와 관련된 첫번째 액션(action)이 실행된 후 스파크는 오류의 파티션을 메모리에 저장하여 후속 연산을 크게 가속화한다.
- 기본 RDD는 RAM에 로드되지 않는다.
- 이는 오류 메시지가 데이터의 작은 부분에 불과하기 때문이다.

위 그림은 모델이 fault tolerance를 달성하는 방법을 설명하기 위한 3개의 쿼리에 대한 RDD의 계보(lineage) 그래프를 보여준다.
이 쿼리에서는 라인에서 필터의 결과인 오류로 시작하여 collect를 실행하기 전에 추가 필터와 맵을 적용했다.
스파크 스케줄러는 나머지 두 개의 변환을 파이프라인화하고 캐시된 오류 파티션을 보유한 노드에 태스크 집합을 전송한다.
또한 오류 파티션이 손실되면 spark는 해당 라인 파티션에만 필터를 적용하여 재구성한다.

2.3 Advantages of the RDD Model

RDD의 coarse-grained 변환 모델은 분산 공유 메모리(DSM) 대비 체크포인트 오버헤드와 장애 복구 비용을 효과적으로 낮춘다.

coarse-grained : fine-grained 보다 큰 구성 요소이다. 하나 이상의 fine-grained 서비스를 더 coarse-grained로 간단히 정리할 수 있다. ¹²
fine-grained : 더 큰 구성 요소를 구성하는 더 작은 구성 요소. 낮은 수준의 서비스 ¹²

DSM(distributed shared memory) 시스템에서 애플리케이션은 전역 주소 공간의 임의의 위치에 읽고 쓴다.
DSM은 매우 일반적인 추상화이지만, 이러한 일반성은 상용 클러스터에서 효율적이고 fault tolerance한 방식으로 구현하기 어렵다.
RDD는 coarse-grained 변환을 통해서만 생성이 되며, 보다 효율적인 fault tolerance를 허용한다.
- 특히, RDD는 계보(lineage)를 사용하여 복구할 수 있기 때문에 체크포인트의 오버헤드를 부담할 필요가 없다.
- 또한, RDD의 손실된 파티션만 장애 시 재계산하면 되며, 전체 프로그램을 롤백(rollback)할 필요 없이 서로 다른 노드에서 병렬로 재계산할 수 있다.

체크포인트 복제 없이 계보만으로 손실된 파티션을 즉시 재연산한다.

RDD는 시스템이 맵리듀스와 같이 느린 태스크의 백업 복사본을 실행함으로써 느린 노드(straggler)를 완화할 수 있는 이점이 있다.
- 백업 작업은 DSM에서는 어려울 수 있는데, 백업 작업의 두 복사본이 동일한 메모리 위치에 액세스하여 서로의 업데이트를 방해하기 때문이다.
- 스트래글러(straggler) : 여러 작업들 간에 작업량이 균일하지 않게 배포되거나, 데이터 스큐(skewed)로 인해 한 작업에서 더 많은 데이터를 처리하는 경우에 발생할 수 있다. ¹³
또한 RDD의 대량(bulk) 연산에서 런타임은 성능을 향상시키기 위해 데이터 인접성(locality)을 기반으로 태스크를 스케줄할 수 있다.
RDD는 스캔 기반 작업에만 사용되는 한 저장할 메모리가 충분하지 않을 때 성능이 저하된다.
- RAM에 맞지 않는 파티션은 디스크에 저장할 수 있으며 현재 데이터 병렬 시스템과 유사한 성능을 제공한다.

2.4 Applications Not Suitable for RDDs

모든 워크로드가 RDD에 적합한 것은 아니며, 세분화된 상태 업데이트가 필요한 비동기 시스템은 여전히 전용 시스템이 유리하다.

RDD는 웹 응용 프로그램이나 증분 웹 크롤러를 위한 스토리지 시스템과 같이 공유(shared) 상태로 비동기적으로(asynchronous) fine-grained하게 업데이트되는 응용 프로그램에는 적합하지 않다.
이러한 애플리케이션의 경우 기존 업데이트 로깅 및 데이터 체크포인트를 수행하는 시스템을 사용하는 것이 더 효율적이다.
spark의 목표는 배치 분석을 위한 효율적인 프로그래밍 모델을 제공하고 이러한 비동기 애플리케이션을 특수 시스템에 맡기는 것이다.

이러한 RDD 설계를 바탕으로 사용자가 클러스터 연산을 어떻게 작성하는지 프로그래밍 인터페이스를 확인한다.

3. Spark Programming Interface

스파크는 스칼라(Scala) 언어 기반으로 고수준 함수형 API를 제공하며, 드라이버와 워커 간의 유기적 분산 실행을 지원한다.

스칼라(scala)는 간결성(conciseness)과 효율성(efficiency)의 조합으로 이 언어를 선택했다.
- 간결성은 대화적(interactive) 사용에 편리함
- 효율성은 정적(static) 타이핑 때문
  - 정적 타이핑(static typing) : 코드를 작성할 때 컴퓨터적 구조를 명시해주기 때문에 컴퓨터가 해야할 일을 덜어 준다. ¹⁴

스파크를 사용하기 위해 개발자는 워커 클러스터에 연결하는 드라이버(driver) 프로그램을 작성한다.
- 워커(worker) : 여러 연산에 걸쳐 RDD 파티션을 RAM에 저장할 수 있는 long-lived 프로세스이다.
드라이버는 하나 이상의 RDD를 정의하고 RDD에 대한 액션(action)을 호출한다.
- RDD의 계보(lineage)도 추적한다.

3.1 RDD Operations in Spark

RDD 연산은 지연 평가되는 변환(Transformation)과 결과를 산출하는 액션(Action)으로 엄격히 구분된다.

Table 2는 스파크에서 사용할 수 있는 주요 RDD 변환 및 액션을 보여준다.
- 변환(transformation)은 새로운 RDD를 정의하는 게으른(lazy) 연산이며, 액션(action)은 프로그램에 값을 반환하거나 외부 스토리지에 데이터를 쓰는(write) 계산을 시작한다.
  - Lazy evaluation : 계산의 결과값이 필요할 때까지 계산을 늦추는 기법. 즉시, 평가하지 않고 필요한 것만 평가하는 방법 ¹⁵

3.2 Example Applications (요약)

반복 연산과 조인(Join) 구조에서 RDD의 인메모리 캐싱과 파티셔닝 제어가 I/O 병목을 완화한다.

로지스틱 회귀와 PageRank의 전체 코드는 상단 2장 예제 섹션을 참고.
포인트: 반복 연산은 persist로 가속, 키 파티셔닝으로 join 셔플 최소화, 좁은 종속성 파이프라이닝으로 스테이지 수 축소.

3.2.1 Logistic Regression

머신러닝 알고리즘은 본질적으로 반복 연산을 수반하므로, 중간 데이터를 메모리에 유지하면 반복마다 발생하는 I/O 오버헤드를 줄여 실행 속도를 높일 수 있다.

val points = spark.textFile(...)
                  .map(parsePoint).persist()
val w = // random initial vector
for (i <- 1 to ITERATIONS) {
  val gradient = points.map{ p =>
   p.x * (1/(1+exp(-p.y*(w dot p.x)))-1)*p.y
  }.reduce((a,b) => a+b)
  w -= gradient
}

3.2 PageRank

PageRank 알고리즘은 각 문서에 연결된 문서의 기여도를 합산하여 각 문서에 대한 순위를 반복적으로(iteratively) 업데이트한다.
각각의 반복에서, 각 문서는 $\frac{r}{n}$의 기여도를 이웃들에게 보낸다.
- $r$ : 순위(rank)
- $n$ : 이웃의 수
그런 다음 순위를 $\alpha/N+(1-\alpha)\sum c_i$으로 업데이트한다.
- 합계(sum) : 자신이 받은 기여도를 초과를 나타낸다.
- N : 문서의 총 수
PageRank를 spark로 나타내면 다음과 같다.

// Load graph as an RDD of (URL, outlinks) pairs
val links = spark.textFile(...)
                 .map(...)
                 .persist()
val ranks = // RDD of (URL, rank) pairs
for (i <- 1 to ITERATIONS) {
  // Build an RDD of (targetURL, float) pairs
  // with the contributions sent by each page
  val contribs = links.join(ranks).flatMap {
    (url, (links, rank)) =>
      links.map(dest => (dest, rank/links.size))
  }
  // Sum contributions by URL and get new ranks
  ranks = contribs.reduceByKey((x,y) => x+y)
                  .mapValues(sum => a/N + (1-a)*sum)
}

이 프로그램은 Figure 3의 RDD 계보 그래프로 이어진다.
각 반복마다 이전 반복의 기여와 순위 및 정적 링크 데이터 셋을 기반으로 새로운 순위 데이터 셋을 생성한다.
링크 데이터 집합의 파티션은 입력 파일의 블록(block)에서 맵을 다시 실행하면 효율적으로 재구성될 수 있으므로 복제할 필요가 없다.
또한 RDD의 파티셔닝을 제어함으로써 PageRank의 통신을 최적화할 수 있다.
- 링크에 대한 파티셔닝을 지정하면 동일한 방식으로 순위를 파티셔닝할 수 있으며, 링크와 순위 간의 join 연산이 통신을 필요로 하지 않도록 보장할 수 있다.

4. Representing RDDs

스파크는 5가지 핵심 메타데이터 인터페이스로 RDD를 단순하게 추상화하여 광범위한 연산자를 통합 스케줄링한다.

RDD는 광범위한 변환에 걸쳐 계보(lineage)를 추적할 수 있는 RDD의 표현(representation)을 선택한다.
RDD를 구현하는 시스템은 가능한 풍부한 변환 연산자 집합을 제공해야 하며 사용자가 임의적인 방법으로 변환 연산자를 구성할 수 있어야 한다.
저자들은 이러한 목표를 용이하게 하는 RDD에 대한 간단한 그래프 기반 표현을 제안한다.
- spark에서 이 표현을 사용하여 각각의 스케줄러에 특별한 논리를 추가하지 않고 광범위한 변환을 지원하여 시스템 설계를 크게 단순화했다.
- 파티션 집합(Partitions) : 데이터셋을 구성하는 원자적 조각 목록
- 부모 종속성 집합(Dependencies) : 부모 RDD와의 의존 관계 목록
- 연산 함수(Iterator) : 부모 데이터셋으로부터 해당 파티션을 계산하는 함수
- 파티셔너 메타데이터(Partitioner) : 키-값 RDD의 해시/범위 파티셔닝 정보
- 기본 위치 정보(Preferred Locations) : 데이터 지역성을 고려한 파티션 블록 위치

다음으로 RDD 간의 종속성을 어떻게 표현하느냐이다.
저자들은 종속성을 2가지 유형으로 분류했다.
부모 RDD의 각 파티션이 자녀 RDD의 최대 한 파티션에 의해 사용되는 좁은 종속성(narrow dependency)과 여러 자녀 파티션이 종속될 수 있는 넓은 종속성(wide dependency)이다.
- 좁은 종속성(narrow dependency) : 부모 RDD 파티션이 자식 RDD 파티션과 최대 1:1로 대응된다. ¹⁶ ¹⁷
  - Shuffling이 거의 발생하지 않으며, 빠르다.
- 넓은 종속성(wide dependency) : 부모 RDD 파티션이 자식 RDD 파티션과 N:1로 대응된다. ¹⁶ ¹⁷
  - Shuffling이 많은 데이터에 발생하며, 느리다.
  - 임의의 데이터만으로 실행할 수는 없으며, 특별한 방법, 예를 들면 키의 값에 따라 파티셔닝된 데이터를 요구한다. ¹⁸
  - 결국 키의 재분포, 즉 셔플이 필요하다.
- 사전에 작업된 RDD를 부모 RDD, 이를 바탕으로 작업된 RDD로 정의하며, 자식 RDD를 만들게 한 map 메소드를 종속성(dependency)이라 한다. ¹⁶

이같은 구별은 2가지 이유로 유용하다.

좁은 종속성은 모든 부모 파티션을 계산할 수 있는 하나의 클러스터 노드에서 파이프라인을 실행할 수 있다.
- 요소별로 맵 다음에 필터를 적용할 수 있다.
- 반면 넓은 종속성을 위해서는 모든 부모 파티션의 데이터를 사용할 수 있어야 하며 맵리듀스와 같은 작업을 사용하여 노드 간에 서로 섞어야(shuffled) 한다.
노드 장애 후 복구는 손실된 부모 파티션만 재계산하면 되고 서로 다른 노드에서 병렬로 재계산할 수 있기 때문에 좁은 종속성으로 더 효율적으로 대처할 수 있다.
- 반면 넓은 종속성의 계보 그래프에서, 하나의 실패한 노드는 RDD의 모든 부모로부터 일부 파티션이 손실되어 완전한 재실행이 필요할 수 있다.
- RDD를 위한 이 공통 인터페이스는 스파크의 대부분의 변환을 20줄 미만의 코드로 구현 가능하다.

아래는 여러 RDD 구현이 요약된 것이다.
HDFS files
- RDD가 HDFS의 파일일 경우, 파티션은 파일의 각 블록에 대해 하나의 파티션(partition)을 반환하고, 기본 설정 위치는 블록이 있는 노드를 제공하며, 반복자(iterator)는 블록을 읽는다.(read)
map
- 임의의 RDD에서 맵을 호출하면 MappedRDD 개체가 반환된다.
- 이 개체는 부모와 동일한 파티션과 기본 위치를 가지지만 반복자 메서드에서 부모 레코드에 매핑하기 위해 전달된 함수를 적용한다.
union
- 두 개의 RDD에서 union을 호출하면 파티션이 부모의 파티션이 union인 RDD가 반환된다.
- 각 자식 파티션은 해당 부모에 대한 좁은 종속성을 통해 계산된다.
join
- 2개의 RDD를 결합하면 2개의 좁은 종속성, 2개의 넓은 종속성 또는 혼합이 발생할 수 있다.
  - 좁은 종속성 : 해시(hash)/범위(range)가 모두 동일한 파티션으로 분할될 경우

5. Implementation

스파크는 메소스(Mesos) 클러스터 관리자 상에서 동작하며, DAG 스케줄링과 대화형 인터프리터 연동을 효율적으로 결합했다.

각 스파크 프로그램은 자체 드라이버(마스터)와 워커가 있는 별도의 애플리케이션으로 실행되며, 클러스터 매니저를 통해 리소스를 할당받고 워커 노드에 태스크를 전달한다.
- Mesos : 데이터 센터에서 리소스를 동적으로 할당하는 것을 목표로 하는 distributed kernel이고 리소스 공유 기능을 사용하는 수많은 프레임워크를 제공한다. ² ¹⁹

5.1 Job Scheduling

스케줄러는 액션 호출 시 계보 그래프를 분석하여 셔플 경계를 기준으로 스테이지(Stage) DAG를 분할한다.

사용자가 RDD에서 액션(ex. count or save)을 실행할 때마다 스케줄러는 Figure 5와 같이 실행할 단계의 DAG를 구축하기 위해 해당 RDD의 계보 그래프를 검사한다.
- 방향성 비순환 그래프(DAG; Directed Acyclic Graph) : 개별 요소들이 특정한 방향을 향하고 있으며, 서로 순환하지 않는 구조로 짜여진 그래프 ²⁰
  
  이미지출처 ⁷
단계(stage)들의 경계는 넓은 종속성에 필요한 셔플 연산 또는 부모 RDD의 계산을 단락(short-circuit)시킬 수 있는 “이미” 계산된 파티션이다.
그런 다음 스케줄러는 타겟 RDD를 계산할 때까지 각 단계에서 누락된(missing) 파티션을 계산하는 태스크를 시작한다.
스케줄러는 지연 스케줄링(delay scheduling)을 사용하여 데이터 인접성(locality)을 기반으로 기계에 작업을 할당한다.
스파크의 모든 계산은 현재 드라이버 프로그램에서 호출된 액션에 응답하여 실행된다.

5.2 Interpreter Integration

스파크는 대화형 쉘 환경을 위해 스칼라 인터프리터의 클래스 전송과 싱글톤 코드 생성을 클러스터 환경에 맞게 개조했다.

스칼라(scala) 인터프리터는 일반적으로 사용자가 입력한 각 라인에 대한 클래스를 컴파일하여 JVM에 적재하고 함수를 호출하는 방식으로 동작한다.
이 클래스는 해당 라인의 변수나 함수를 포함하고 초기화 방법으로 라인의 코드를 실행하는 싱글톤(singleton) 객체를 포함한다.
- 싱글톤 패턴(Singleton Pattern) : 객체의 인스턴스가 오직 1개만 생성되는 패턴을 의미한다.

스파크에서 인터프리터를 2가지 변경했다.

클래스 배송(Class shipping) : 워커 노드가 각 라인에 생성된 클래스에 대한 바이트 코드를 가져올 수 있도록 인터프리터가 HTTP를 통해 이러한 클래스를 서비스하도록 했다.
수정된 코드 생성(Modified code generation) : 일반적으로 코드의 각 라인에 대해 생성된 싱글톤 객체는 해당 클래스의 정적 메서드를 통해 접근한다.

5.3 Memory Management

스파크는 메모리와 CPU 비용 간의 트레이드오프를 맞추기 위해 3가지 RDD 저장 옵션과 LRU 교체 정책을 제공한다.

스파크는 persistent RDD를 저장하기 위한 3가지 옵션(option)을 제공한다.
- 직렬화(serialization) : 객체를 직렬화하여 네트워크 상으로 전송 가능한 형태로 만드는 것을 의미한다. 객체들의 데이터를 연속적인 데이터로 변형하여 stream을 통해 데이터를 읽도록 한다. ²¹
  - 객체들을 통째로 파일로 저장하거나 전송하고 싶을 때 주로 사용된다.
- 역직렬화(deserialization) : 직렬화된 파일 등을 역으로 직렬화하여 다시 객체의 형태로 만드는 것을 의미한다. ²¹
  - 저장된 파일 읽거나 전송된 스트림 데이터를 읽어 원래 객체의 형태로 복원한다.
  이미지출처 ⁶

역직렬화된(deserialized) 자바 객체로서의 인메모리(in-memory) 스토리지
- 인메모리 데이터베이스(in-memory database) : 데이터 스토리지의 메인 메모리에 설치되어 운영되는 방식의 데이터베이스 관리 시스템 ²²
  - 인메모리 방식은 메모리상에 색인을 넣어 필요한 모든 정보를 메모리상의 색인을 통해 빠르게 검색할 수 있다.
직렬화된(serialized) 데이터로서의 인메모리 스토리지
디스크 상의 스토리지

첫 번째 옵션은 JVM이 기본적으로 각 RDD 요소에 액세스할 수 있기 때문에 가장 빠른 성능을 제공한다.
두 번째 옵션은 공간(space)이 제한될 때 성능 저하를 감수하고 자바 객체 그래프보다 메모리 효율적인 표현을 선택할 수 있도록 한다.
세 번째 옵션은 RAM에 보관하기에는 너무 크지만 매번 사용할 때마다 재계산하는 데 비용이 많이 드는 RDD에 유용하다.
사용 가능한 제한된 메모리를 관리하기 위해 RDD 수준의 LRU(Least Recently Used) 제거 정책을 사용한다.
- LRU(Least Recently Used) : 가장 오랫동안 참조되지 않은 페이지를 교체하는 방식

5.4 Support for Checkpointing

계보가 지나치게 길어지는 경우를 대비해 스파크는 안정적인 스토리지로의 명시적 체크포인트 API를 병행 지원한다.

장애(failure)가 발생한 후 RDD를 복구(recovery)하는 데 항상 계보(lineage)를 사용할 수 있지만, 긴 계보 체인을 가진 RDD의 경우 이런 복구는 시간이 많이 걸릴 수 있다.
따라서 일부 RDD를 안정적인 스토리지로 체크포인트하는 것이 도움이 될 수 있다.
스파크는 현재 체크포인트를 위한 API를 제공하지만, 체크포인트할 데이터의 결정은 사용자에게 맡긴다.
- 좁은 종속성과 넓은 종속성에 따라 계산 차이가 다르다.

이처럼 잘 설계된 실행 엔진과 메모리 모델이 실제 벤치마크에서 어떤 성능을 보여주는지 살펴본다.

6. Evaluation

실험 결과, 스파크는 반복 연산에서 하둡 대비 최대 20배 향상된 성능과 초 단위 장애 복구 능력을 입증했다.

결과는 다음과 같다.
- 스파크는 반복적인 기계 학습 및 그래프 애플리케이션에서 하둡을 최대 20배 능가한다.
  - 데이터를 자바 객체로 메모리에 저장함으로써 입출력 및 역직렬화 비용을 피할 수 있기 때문에 속도가 향상된다.
- 사용자가 작성한 애플리케이션은 성능과 확장성이 뛰어나다.
- 노드에 장애가 발생하면 스파크는 손실된 RDD 파티션만 재구성하여 신속하게 복구할 수 있다.
- 스파크는 1TB 데이터 세트를 5-7초의 지연시간(latency)으로 대화식(interactively)으로 쿼리하는 데 사용할 수 있다.

6.1 Iterative Machine Learning Applications

반복 머신러닝 연산에서 RDD 인메모리 유지는 하둡의 고질적인 프레임워크 오버헤드와 I/O 병목을 제거한다.

RDD를 통해 데이터를 공유하면 향후 반복 속도가 크게 빨라진다는 것을 알 수 있다.
하둡은 여러 요인으로 인해 여전히 속도가 느렸다.

하둡 소프트웨어 스택(stack)의 최소 오버헤드
- 스택(stack) : 제한적으로 접근할 수 있는 나열 구조 ²³
- 잡 설정, 작업 시작, 정리 등의 최소 요구 사항을 완료하는 데 최소 25초 이상의 오버헤드가 발생했다.
데이터를 서빙할 때 HDFS의 오버헤드
- HDFS는 각 블록에 서비스를 제공하기 위해 여러 개의 메모리 복사본과 체크섬을 수행했다.
이진 레코드를 사용 가능한 메모리 내 자바 객체로 변환하기 위한 역직렬화 비용

RDD 요소를 메모리에 자바 객체로 직접 저장함으로써 스파크는 이러한 모든 오버헤드를 피할 수 있다.

6.2 PageRank

PageRank와 같은 그래프 알고리즘에서도 파티셔닝 보존과 인메모리 캐싱을 통해 하둡 대비 빠른 실행 속도를 보인다.

6.3 Fault Recovery

장애 발생 시 100GB 단위의 데이터를 복제할 필요 없이, 10KB 미만의 계보 그래프로 손실 파티션만 병렬 복구한다.

체크포인트 기반 장애 복구 메커니즘을 사용하는 경우 복구는 체크포인트 빈도에 따라 최소 여러 번 반복을 재실행해야 할 수 있다.
게다가, 시스템은 100GB의 작업 세트(이진수로 변환된 텍스트 입력 데이터)를 네트워크를 통해 복제해야 하며, RAM에 복제하기 위해 스파크의 두 배 메모리를 사용하거나 디스크에 100GB를 쓰기 위해 기다려야 한다.
하지만 RDD에 대한 계보(lineage) 그래프는 모두 크기가 10KB 미만이었다.

10KB 미만의 계보 그래프 메타데이터로 분산 체크포인트 및 복제 부담을 경감했다.

7. Discussion (한계 및 트레이드오프)

RDD는 조잡한(coarse-grained) 변환만 지원하므로, 다수의 스레드가 공유 상태를 세밀하게 갱신해야 하는 응용에는 적합하지 않다. ²
영속 데이터를 메모리에 유지하려면 클러스터 메모리가 충분해야 하며, 메모리가 부족해 디스크로 내려가면 이점이 줄어든다. ²
계보(lineage) 기반 복구는 재계산 비용과 저장 비용의 맞바꿈이다. 계보가 길어지면 복구 시간이 늘어나므로, 넓은 의존성 구간에는 체크포인트를 두는 편이 낫다. ²
짧은 태스크를 다수 생성하는 구조상 스트래글러가 생기기 쉬워, 백업 태스크 실행으로 지연을 줄이는 대신 자원을 추가로 사용한다. ²

핵심 요약 및 시사점

계보 기반 결함 감내 (Lineage-driven Fault Tolerance): 데이터 복제나 잦은 체크포인트 대신 coarse-grained 변환 계보를 추적하여, 파티션 장애 시 손실된 조각만 결정적으로 재계산하는 경량 무복제 결함 감내를 실현했다.
효율적인 인메모리 데이터 재사용: 반복 머신러닝 및 대화형 쿼리에서 중간 결과를 외부 분산 파일 시스템(HDFS)에 쓰지 않고 RAM에 유지하여, 디스크 I/O와 직렬화/역직렬화 오버헤드를 제거하고 최대 20배 이상의 성능 향상을 입증했다.
종속성 분류와 지능적 DAG 스케줄링: RDD 의존성을 파이프라이닝 가능한 Narrow Dependency와 네트워크 셔플이 필요한 Wide Dependency로 명확히 이원화하여 효율적인 실행 스테이지 경계를 수립했다.
간결하고 유연한 통합 인터페이스: 파티션, 부모 의존성, 연산 함수 등 5가지 메타데이터 인터페이스만으로 다양한 병렬 연산자를 20줄 미만의 코드로 확장 가능하게 단순화했다.

References

KakaoTalk Twitter Facebook LinkedIn