CHAPTER 4. 구조적 API 개요

Note 개념 복습
스파크는 트랜스포메이션의 처리과정을 정의하는 분산 프로그래밍 모델 다수의 트랜스포메이션 → 지향성 비순환 그래프(DAG)로 표현되는 명령을 생성 액션은 하나의 Job을 클러스터에서 실행하기 위해 스테이지와 task로 나누고 DAG처리 프로세스 실행 DataFrame, Dataset : 트랜스포메이션과 액션을 다루는 논리적 구조 DataFrame이나 Dataset을 만들려면 트랜스포메이션을 호출 연산을 시작하거나 사용한 언어에 맞는 데이터 타입으로 변환하려면 액션 호출

4.0 구조적 API의 특징

• 데이터 흐름을 정의하는 기본 추상화 개념
• 구조적 API가 처리하는 데이터 유형
  • 비정형 로그파일
  • 반정형 CSV파일
  • 파케이(Parquet) 파일
• 구조적API의 분산 컬렉션API
  • Dataset
  • DataFrame
  • SQL 테이블과 뷰
• 배치와 스트리밍처리에서 구조적 API를 사용할 수 있음
  • 배치작업 ↔ 스트리밍 작업 : 손쉽게 변환 가능

4.1 DataFrame과 Dataset

• 잘 정의된 로우와 컬럼을 가지는 분산 테이블 형태의 컬렉션

• 각 컬럼은 다른 컬럼과 동일한 로우를 가져야함

• 모든 로우는 같은 데이터 타입 정보를 가지고 있어야함

• 지연연산의 실행계획이며 불변성을 가진다.

  • DataFrame에 액션을 호출하면 스파크는 트랜스포메이션을 실제로 실행하고 결과를 반환한다.

    이는 사용자가 원하는 결과를 얻기위해 로우와 컬럼을 처리하는 방법에 대한 계획을 나타낸다.

4.2 스키마

• 분산 컬렉션에 저장할 데이터 타입을 정의하는 방법
• DataFrame의 컬럼명과 데이터 타입을 정의
• 스키마 정의 방법
  • 데이터소스에서 얻기(스키마 온 리드 schema-on-read)

print(df.schema)

• 직접 정의

from pyspark.sql.types import StructField, StructType, StringType, LongType

myManualSchema = StructType([
      StructField("DEST_COUNTRY_NAME", StringType(), True),
      StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
      StructField("count", LongType(), True)
])
print(myManualSchema)

• 스키마는 여러 데이터 타입으로 구성되기 때문에 어떤 데이터 타입이 어느 위치에 있는지 정의하는 방법 필요

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

• 카탈리스트 엔진
  • 실행 계획 수립과 처리에 사용하는 자체 데이터 타입 정보를 가짐
  • 다양한 실행 최적화 기능 제공

• 자체 데이터 타입을 지원하는 여러 언어 API와 직접 매핑, 각 언어에 대한 매핑 테이블을 가지고 있음

  • 파이썬이나 R을 이용해 스파크의 구조적 API를 사용하더라도 대부분 연산은 파이썬이나 R의 데이터 타입이 아닌 스파크의 데이터 타입을 사용
  • 스파크가 지원하는 언어를 이용해 작성된 표현식을 카탈리스트 엔진에서 스파크의 데이터 타입으로 변환해 명령을 처리

# 파이썬으로 작성된 스파크의 덧셈 연산을 수행하는 코드
df = spark.range(500).toDF("number")
df.select(df["number"] + 10)

4.3.1 DataFrame과 Dataset비교

• DataFrame
  • 비타입형
  • 스키마에 명시된 데이터 타입의 일치 여부를 런타임이 되어야 확인
  • row타입으로 구성된 Dataset
    • row타입 : 스파크가 사용하는 연산에 최적화된 인메모리 포맷의 내부적인 표현 방식
    • 가비지 컬렉션과 객체 초기화부하가 있는 JVM 데이터 타입을 사용하는 대신 자체 데이터 포맷을 사용하기 때문에 효율적인 연산 가능
  • DataFrame을 사용하면 스파크의 최적화된 내부 포맷을 사용할 수 있음
    • 파이썬이나 R에선 Dataset은 사용불가능 하지만 최적화된 포맷인 DataFrame으로 처리 가능
    • 스파크가 지원하는 어떤언어 API를 사용하더라도 동일한 효과와 효율성을 얻을 수 있음
• Dataset
  • 타입형
  • 스키마에 명시된 데이터 타입의 일치 여부를 컴파일 타임에 확인
  • JVM기반의 언어인 스칼라와 자바에서만 지원
    • 스칼라의 케이스 클래스나 자바 빈을 사용함

4.3.2 컬럼

컬럼의 표현 세가지

• 정수형이나 문자열 같은 단순 데이터 타입
• 배열이나 맵같은 복합 데이터 타입
• NULL 값

스파크는 데이터 타입의 모든 정보를 추적하며 다양한 컬럼 변환 방법 제공

스파크의 컬럼은 테이블의 컬럼으로 생각할 수 있음

4.3.3 로우

• 데이터 레코드

• DataFrame의 레코드는 ROW타입으로 구성

• 로우는 SQL, RDD, 데이터소스에서 얻거나 직접 만들 수 있음

#로우 객체로 이루어진 배열 반환
spark.range(2).collect()

4.3.4 스파크의 데이터 타입

스파크는 여러 가지 내부 데이터 타입을 갖고 있음

다양한 프로그래미 언어의 데이터 타입이 스파크의 어떤 타입과 매핑되는지 표를 통해 확인 가능

스파크의 데이터 타입을 파이썬에서 사용할 경우의 코드

from pyspark.sql.types import *

b = ByteType()

4.4 구조적 API의 실행 과정

1. DataFrame/Dataset/SQL을 이용해 코드 작성
   작성한 코드는 콘솔이나 spark-submit 셸 스크립트로 실행
2. 정상적인 코드라면 스파크카 논리적 실행 계획으로 변환
3. 스파크가 논리적 실행 계획을 물리적 실행계획으로 변환하며 그 과정에서 추가적인 최적화를 할 수 있는지 확인
4. 스파크가 클러스터에서 물리적 실행 계획 (RDD처리)을 실행

4.4.1 논리적 실행 계획

• 사용자 코드를 논리적 실행 계획으로 변환
• 추상적 트랜스포메이션만 표현함
• 드라이버나 익스큐터의 정보를 고려하지 않음

1. 사용자코드 → 검증 전 논리적 실행 계획

   • 사용자 코드를 논리적 실행 계획으로 변환
   • 추상적 트랜스포메이션만 표현함
   • 드라이버나 익스큐터의 정보를 고려하지 않음
   • 코드의 유효성과 테이블이나 컬럼의 존재 여부만을 판단 → 실행계획을 검증하지 않은 상태
   • 사용자의 다양한 표현식을 최적화된 버전으로 변환

2. 스파크 분석기(Analyzer)

   • 컬럼과 테이블을 검증하기 위해 카탈로그/모든 테이블의 저장소/DataFrame 정보를 활용
   • 필요한 테이블이나 컬럼이 카탈로그에 없다면 검증 전 논리적 실행 계획이 만들어지지 않음

3. 논리적 최적화(카탈리스트 옵티마이저)

• 조건절 푸시 다운(predicate pushing down)이 선택절 구문을 이용해 논리적 실행계획을 최적화하는 규칙의 모음

• 도메인에 최적화된 규칙을 적용할 수 있는 카탈리스트 옵티마이저의 확장형 패키지 생성 가능

4.4.2 물리적 실행 계획

• (== 스파크 실행계획)
• 논리적 실행 계획을 클러스터 환경에서 실행하는 방법 정의
• 다양한 물리적 실행 전략을 생성 → 비용 모델을 이용해서 비교 → 최적의 전략을 선택
  • 비용 비교 예시) 사용하려는 테이블의 크기나 파티션 수 등의 물리적 속성을 고려해 지정된 조인 연산 수행에 필요한 비용을 계산하고 비교
• 스파크는 DataFrame, Dataset, SQL로 정의된 쿼리를 RDD 트랜스포메이션으로 컴파일 → 스파크를 컴파일러로 부르기도 함

4.4.3 실행

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