Spark가 대체 뭐야? (3)

Posted on | In

.

구조적 API

구조적 API에는 3가지의 분산 컬렉션 API가 존재

  • Dataset
  • DataFrame
  • SQL 테이블과 뷰

배치와 스트리밍 처리에 구조적 API 사용 가능.

→ 구조적 API를 활용한다면 배치 작업을 스트리밍 작업으로 변환 가능.

  • 구조적 API는 데이터 흐름을 정의하는 기본 추상화 개념.
    • 타입형/비타입형 API의 개념과 차이점
    • 핵심 용어
    • 스파크가 구조적 API의 데이터 흐름을 해석하고 클러스터에서 실행하는 방식

4.1 DataFrame, Dataset

  • 잘 정의된 row, column을 가지는 분산 테이블 형태의 컬렉션
  • 결과를 생성하기 위해 어떤 데이터에 어떤 연산을 적용해야하는지 정의하는 지연 연산의 실행계획
    • 불변성을 가짐

4.2 스키마

  • DataFrame의 컬럼명과 데이터 타입을 정의
  • schema-on-read 혹은 직접 정의

4.3 스파크의 구조적 데이터 타입 개요

  • 스파크 카탈리스트 엔진 → 실행 최적화 기능
1
2
val df = spark.range(500).toDF("number")
df.select(df.col("number")+10)
  • 스파크는 사실상 프로그래밍 언어

4.3.1 DataFrame, Dataset 비교

  • DataFrame → 비타입형
    • 스키마에 명시된 데이터 타입의 일치여부를 런타임에 확인
    • Row타입으로 구성된 Dataset
      • Row 타입은 스파크가 사용하는 연산에 최적화된 인메모리 포맷
      • JVM 데이터 타입은 GC와 객체 초기화 부하가 있지만, Row타입은 이런 점에서 자유로움
    • DataFrame을 사용하면 스파크의 최적화된 내부 포맷을 사용할 수 있음.
  • Dataset → 타입형
    • 스키마에 명시된 데이터 타입의 일치여부를 컴파일 타임에 확인
      • JVM 기반의 scala, java에서만 지원 (scala: case class)

4.3.2 컬럼

  • 정수형이나 문자열 → 단순 데이터 타입
  • 배열, 맵 → 복합 데이터 타입
  • null

4.3.3 row

  • 데이터 레코드
  • DataFrame의 레코드는 Row타입으로 구성
1
spark.range(2).toDF().collect() // Row 객체로 이루어진 배열 반환

4.3.4 스파크 데이터 타입

1
2
3
4
// 스파크 데이터 타입을 스칼라에서 사용하는 방법
import org.apache.spark.sql.types._

val b = ByteType

4.4 구조적 API의 실행 과정

  1. DataFrame/Dataset/SQL을 이용해 코드를 작성
  2. 정상적인 코드라면 스파크가 논리적 실행 계획으로 변환
  3. 스파크는 논리적 실행 계획을 물리적 실행 계획으로 변환 → 추가적인 최적화가 가능한지 확인
  4. 클러스터에서 물리적 실행 계획 (RDD 처리)을 실행

4.4.1 논리적 실행 계획

  • 사용자 코드 → unresolved logical plan
    • 코도의 유효성과 테이블이나 컬럼의 존재 여부만을 판단
  • Analyzer는 컬럼과 테이블을 검증하기 위해 카탈로그, 모든 테이블의 저장소, DataFrame정보를 활용

4.4.2 물리적 실행 계획

  • 스파크 실행 계획 = 물리적 실행 계획
  • 논리적 실행계획을 클러스터 환경에서 실행하는 방법을 정의
  • 다양한 물리적 실행 전략을 생성 → 비용모델을 이용해서 비교 → 전략 선택
  • 물리적 실행 계획 → 일련의 RDD와 트랜스포메이션으로 변환
  • 스파크는 DataFrame, Dataset, SQL로 정의된 쿼리를 RDD 트랜스포메이션으로 컴파일해서 스파크를 컴파일러라고 부르기도 함.

4.4.3 실행

  • 스파크는 물리적 실행 계획을 선정한 다음 RDD(저수준 프로그래밍 인터페이스)를 대상으로 모든 코드를 실행
  • 스파크는 런타임에 전체 테스크나 스테이지를 제거할 수 잇는 자바 바이트 코드를 생성해 추가적인 최적화 수행.

.

4부에서 계속