Spark

트랜스포메이션: narrow와 wide dependency

Spark에서 핵심 데이터 구조는 불변(immutable)이다 — 한 번 만들어지면 바꿀 수 없다. 그렇다면 DataFrame을 '변경'하려면 어떻게 해야 할까? 방법은, 지금 가진 DataFrame을 원하는 DataFrame으로 어떻게 바꿀지 Spark에게 지시하는 것이다. 이 지시가 바로 트랜스포메이션이다.

예를 들어 짝수만 걸러내는 트랜스포메이션은 다음과 같다.

val divisBy2 = myRange.where("number % 2 = 0")

divisBy2 = myRange.where("number % 2 = 0")

이 코드를 실행해도 아무 결과가 출력되지 않는다. 추상적인 트랜스포메이션 하나만 지정했을 뿐이고, 액션이 호출되기 전까지 Spark는 트랜스포메이션에 대해 아무 동작도 하지 않기 때문이다.

트랜스포메이션에는 narrow dependency와 wide dependency 두 종류가 있다. narrow dependency는 각 입력 파티션이 오직 하나의 출력 파티션에만 기여하는 트랜스포메이션이다. 방금 본 where 절이 바로 narrow dependency의 예로, 파티션 하나가 최대 한 개의 출력 파티션에만 기여한다. 반면 wide dependency 트랜스포메이션은 입력 파티션들이 여러 출력 파티션에 걸쳐 기여하며, 이를 셔플(shuffle)이라 부른다 — Spark가 클러스터 전역에 걸쳐 파티션을 주고받는 것이다.

Spark는 narrow dependency에 대해서는 파이프라이닝(pipelining)이라는 최적화를 자동으로 수행한다. 예컨대 DataFrame에 필터를 여러 번 연달아 지정하면 이들은 모두 메모리 안에서 수행된다. 하지만 셔플은 그렇지 않다 — 셔플을 수행할 때 Spark는 결과를 디스크에 기록한다. 셔플 최적화가 흔히 중요한 주제로 다뤄지는 이유가 여기에 있다.

지연 평가(Lazy Evaluation)

지연 평가란 Spark가 트랜스포메이션 실행을 최대한 마지막 순간까지 미루는 것을 의미한다. 데이터를 즉시 수정하는 대신, Spark는 원본 데이터에 적용하고 싶은 트랜스포메이션들의 계획(plan)을 차곡차곡 쌓아 올린다. 그리고 실행을 마지막까지 미룸으로써, 사용자가 작성한 원시(raw) DataFrame 트랜스포메이션들을 클러스터 전체에서 최대한 효율적으로 동작하는 물리 실행 계획으로 컴파일한다. 이는 Spark가 데이터 흐름 전체를 처음부터 끝까지 최적화할 수 있게 해주므로 사용자에게 큰 이점을 준다.

대표적인 예가 '프리디킷 푸시다운(predicate pushdown)'이다. narrow dependency로만 이루어진 대규모 Spark job을 구성했는데 맨 마지막에 원본 데이터에서 단 한 행만 필요하도록 만드는 필터를 지정했다면, 가장 효율적인 실행 방법은 필요한 그 한 행만 바로 접근하는 것이다. Spark는 필터를 자동으로 앞단(원본 데이터 접근 시점)까지 밀어내어 이를 최적화해준다.

액션과 Spark UI

트랜스포메이션은 논리적인 변환 계획을 쌓아 올리는 역할만 한다. 실제 계산을 촉발하려면 액션을 실행해야 한다. 액션은 일련의 트랜스포메이션으로부터 결과를 계산하라고 Spark에게 지시하는 것이다. 가장 단순한 액션은 count로, DataFrame의 전체 레코드 수를 알려준다.

divisBy2.count()

이 코드를 실행하면 비로소 결과가 나온다(0~999 중 짝수는 500개). 액션에는 count 외에도 세 가지 종류가 있다 — 콘솔에 데이터를 보여주는 액션, 데이터를 각 언어의 네이티브 객체로 모으는(collect) 액션, 그리고 결과를 출력 데이터 소스에 쓰는 액션이다.

앞의 count 예에서는 액션을 지정함으로써 Spark job이 시작되어, 먼저 필터 트랜스포메이션(narrow)이 실행되고, 이어서 파티션별로 개수를 세는 집계(wide) 연산이 수행된 뒤, 마지막으로 collect가 그 결과를 해당 언어의 네이티브 객체로 가져온다. 이 전 과정은 Spark에 내장된 모니터링 도구인 Spark UI를 통해 확인할 수 있다.

Spark UI는 드라이버 노드의 4040번 포트에서 제공되며, 로컬 모드라면 http://localhost:4040 으로 접속하면 된다. Spark UI는 Spark job, 환경, 클러스터 상태에 대한 정보를 유지하며 튜닝과 디버깅에 특히 유용하다. 앞의 count 예제의 경우 Spark UI에서는 1개의 Spark job이 2개의 stage와 9개의 task로 실행되었음을 확인할 수 있다.

Spark job은 하나의 개별 액션에 의해 촉발된 트랜스포메이션들의 집합을 의미하며, 그 실행 상태를 Spark UI로 모니터링할 수 있다.