Spark

Spark 툴셋 한눈에 보기

Chapter 2에서 다룬 DataFrame과 SQL은 Spark의 Structured API 중 핵심에 해당한다. 이번 장은 그 핵심 바깥으로 시야를 넓혀, Structured Streaming, MLlib·그래프 분석 같은 고급 분석, RDD·분산 변수 같은 저수준 API, 그리고 커뮤니티가 만든 패키지 생태계까지 Spark가 제공하는 나머지 도구들을 훑어본다.

이 장의 목적은 각 주제를 깊이 파고드는 것이 아니라, 이후 어떤 파트를 먼저 읽어도 되는지 감을 잡을 수 있도록 전체 지도를 보여주는 데 있다.

Dataset: JVM 전용 타입 안전 컬렉션

DataFrame은 Row 타입 객체의 분산 컬렉션이다. 반면 Dataset은 JVM 사용자(Scala/Java)가 case class나 Java bean으로 자신만의 객체 타입을 정의하고, 이를 로컬 컬렉션 다루듯 함수형 프로그래밍 방식(map, filter 등)으로 다룰 수 있게 해주는 API다.

Dataset의 강점은 '필요할 때만' 선택적으로 쓸 수 있다는 점이다. 임의의 case class 객체에 map/filter로 원하는 만큼 로직을 적용한 뒤, 그 결과를 다시 DataFrame으로 되돌려 Spark가 제공하는 수백 개의 내장 함수와 SQL로 이어서 처리할 수 있다. 즉 비즈니스 로직이 복잡해 타입 안전성이 필요한 구간은 낮은 레벨의 Dataset으로 내려가고, 빠른 분석이 필요한 구간은 다시 SQL/DataFrame 레벨로 올라오는 식으로 자유롭게 넘나들 수 있다.

case class ValueAndDouble(value: Long, valueDoubled: Long)

spark.range(2000)
  .map(value => ValueAndDouble(value, value * 2))
  .filter(vAndD => vAndD.valueDoubled % 2 == 0)
  .where("value % 3 = 0")
  .count()

캐싱으로 반복 연산 비용 줄이기

Chapter 2에서 다룬 것처럼 wide transformation은 셔플을 일으키고 Spark는 필요시 디스크로 스필한다. 그런데 같은 데이터 흐름 안에서 하나의 DataFrame을 여러 번 참조해야 하는 경우가 흔하다. 예를 들어 원본 CSV를 읽어 만든 DF1을 바탕으로 목적지별 집계(DF2), 출발지별 집계(DF3), 카운트별 집계(DF4)를 각각 만든다면, 세 DataFrame 모두 같은 부모(DF1)를 공유하지만 지연 실행 특성상 각자 그 부모를 처음부터 다시 계산하게 된다. 책의 로컬 환경 테스트에서 이 과정은 약 2.75초가 걸렸다.

cache()를 호출하면 최초로 DataFrame을 읽어들일 때 그 결과를 메모리에 저장해 두고, 이후 다른 쿼리들은 원본 파일이 아니라 메모리에 저장된 버전을 참조하게 된다. 중요한 점은 DataFrame의 진짜 origin은 항상 안정적인 데이터 소스(파일, 테이블 등)라는 것이고, 캐싱은 그 계보 중간에 새로운 origin을 하나 추가하는 것일 뿐이라는 점이다. 같은 예제를 cache() 적용 후 다시 실행하면 소요 시간이 절반 이하로 줄었다. 데이터셋이 크거나 생성 비용이 큰 연산일수록, 그리고 같은 데이터를 반복 접근하는 반복적 머신러닝 워크로드일수록 캐싱의 이득은 커진다.

데이터가 메모리에 다 담기지 않을 만큼 클 수도 있는데, 이런 경우를 위해 persist() 메서드와 디스크 저장을 포함한 다양한 저장 레벨 옵션이 제공된다(세부 튜닝은 책의 Part IV에서 다룬다).