정현석

개인 저장공간입니다.
Since 2021.10

3 분 소요

[Spark The Definitive Guide - BIG DATA PROCESSING MADE SIMPLE] 책을 중심으로 스파크를 개인 공부를 위해 요약 및 정리해보았습니다. 다소 복잡한 설치 과정은 도커에 미리 이미지를 업로드해 놓았습니다. 즉, 도커 이미지를 pull하면 바로 스파크를 사용할 수 있습니다.

예제에 대한 실행 언어는 파이썬으로 했습니다. 스칼라는 추후에 다루겠습니다.

기본 실행 방법

  1. 예제에 사용 될 데이터들은 도커 이미지 생성 후 spark-3.3.0 안의 하위 폴더 data를 생성 후, 이 폴더에 추가합니다.
    • 데이터와 도커 설치 및 활용하는 방법은 위에 링크를 남겼습니다.
  2. 프로그램 시작은 cd spark-3.3.0 후, ./bin/pyspark 명령어를 실행하시면 됩니다.

1. 개요

  • 이전 글에서는 스파크의 전반적인 내용을 다뤘다.
  • 이번 글에서는 스파크 내부에서 어떤 일이 일어나는지 단계별로 살펴본다.

1.1 종합 예제

  • 미국 교통통계국의 항공운항 데이터 중 일부를 스파크로 분석한 예제를 본다.
  • head data/flight-data/csv/2015-summary.csv 명령어를 통해 파일을 확인한다.

  • 실행결과

    image

  • 스파크는 다양한 데이터소스를 지원한다.

  • 데이터는 SparkSession의 DataFrameReader 클래스를 사용해서 읽는다.

    from pyspark.sql import SparkSession

spark = SparkSession.builder \
  .appName('DataFrameReader') \
  .master("local") \
  .config("spark.some.config.option", "some-value") \
  .getOrCreate() # spark 초기화

SparkSession을 생성하면 스파크 코드를 실행할 수 있다. 그리고 모든 저수준 API, 기존 컨텍스트 그리고 관련 설정 정보에 접근할 수 있다. SparkContext를 이용하면 RDD같은 스파크의 저수준 API를 사용할 수 있다. 모든 스파크 코드는 RDD 명령으로 컴파일된다.

  • 이때 특정 파일 포맷과 몇 가지 옵션을 함께 설정한다.
    • .option("inferSchema", "true") : 스파크 DataFrame의 스키마 정보를 알아내는 스키마 추론(schema inference) 기능을 사용한다. 추론은 파일의 실제 데이터를 읽어 타입을 정하므로, 파일이 클수록 추론에 드는 비용도 커진다. 1
      • 스키마(schema) : 컬럼과 컬럼의 타입을 정의한 목록 2
    • .option("header", "true") : 파일의 첫 로우를 헤더로 지정하는 옵션이다.

    • csv 파일 경로를 불러올 때 경로를 찾지 못하겠다면 다음 코드를 실행해본다.

        import os

  print(os.getcwd())

flightData2015 = spark \
    .read \
    .option("inferSchema", "true") \
    .option("header", "true") \
    .csv("./spark-3.3.0/data/flight-data/csv/2015-summary.csv")
  • 스칼라와 파이썬에서 사용하는 DataFrame은 불특정 다수의 로우와 컬럼을 가진다.
  • 로우의 수를 알 수 없는 이유는 데이터를 읽는 과정이 지연 연산 형태의 트랜스포메이션이기 때문이다.
  • 스파크는 각 컬럼의 데이터 타입을 추론하기 위해 적은 양의 데이터를 읽는다.
  • Figure 2.7은 DataFrame에서 CSV 파일을 읽어 로컬 배열이나 리스트 형태로 변환하는 과정이다.

image

  • DataFrame의 take 액션을 호출하면 이전의 head 명령과 같은 결과를 확인할 수 있다.

  • 실행결과

    image

  • 이제 트랜스포메이션을 추가로 지정한다. 정수 데이터 타입인 count 컬럼을 기준으로 데이터를 정렬한다.
    • sort 메서드는 트랜스포메이션이기 때문에 데이터에 아무런 변화도 일어나지 않는다.
    • explain() : DataFrame의 계보(lineage)나 스파크의 쿼리 실행 계획을 확인할 수 있다. 논리 계획과 물리 계획을 콘솔에 출력해 디버깅에 사용한다. 3
      • 스파크는 실행 계획을 만들고 검토하여 클러스터에서 처리할 방법을 알아낸다.
    • 실행 계획(plan)은 위에서 아래 방향으로 읽으며 최종 결과는 가장 위에, 데이터소스는 가장 아래에 있다.

image

flightData2015.sort("count").explain()

  • 실행결과

    image

  • 이제 트랜스포메이션 실행 계획을 시작하기 위해 액션을 호출한다.

    • 스파크는 셔플 수행 시 기본적으로 200개의 셔플 파티션을 생성한다. 이 값은 spark.sql.shuffle.partitions의 기본값 200으로, 조인이나 집계를 위한 셔플에 적용된다. 4
    • 이 값을 5로 설정해 셔플의 출력 파티션 수를 줄인다.

image

spark.conf.set("spark.sql.shuffle.partitions", "5")

flightData2015.sort("count").take(2)
  • 트랜스포메이션의 논리적 실행 계획은 DataFrame의 계보를 정의한다.
  • 스파크는 계보를 통해 입력 데이터에 수행한 연산을 전체 파티션에서 어떻게 재연산하는지 알 수 있다.

계보는 스파크의 프로그래밍 모델인 함수형 프로그래밍의 핵심이다. 함수형 프로그래밍은 데이터의 변환 규칙이 일정한 경우 같은 입력에 대해 항상 같은 출력을 생성한다는 의미이다.

  • 사용자는 물리적 데이터를 직접 다루지 않는다. 대신 앞서 설정한 셔플 파티션 파라미터와 같은 속성으로 물리적 실행 특성을 제어한다.

1.2 DataFrame과 SQL

  • 스파크는 언어에 상관없이 같은 방식으로 트랜스포메이션을 실행할 수 있다.
  • 사용자가 SQL이나 DataFrame으로 비즈니스 로직을 표현하면 스파크에서 실제 코드를 실행하기 전에 그 로직을 기본 실행 계획(plan)으로 컴파일한다.
  • 스파크 SQL을 사용하면 모든 DataFrame을 테이블이나 뷰로 등록한 후 SQL 쿼리를 사용할 수 있다.
    • createOrReplaceTempView 메서드를 호출하면 모든 DataFrame을 테이블이나 뷰로 만들 수 있다. 이때 만든 임시 뷰의 수명은 이를 만든 SparkSession에 묶인다. 5
flightData2015.createOrReplaceTempView("flight_data_2015")
  • 이제 SQL로 데이터를 조회할 수 있다.
    • spark.sql : 새로운 DataFrame을 반환하는 메서드로 SQL 쿼리를 실행한다.
      • spark는 SparkSession의 변수이다.
    • sqlWay와 dataFrameWay 모두 동일한 기본 실행 계획으로 컴파일된다.
sqlWay = spark.sql("""
SELECT DEST_COUNTRY_NAME, count(1)
FROM flight_data_2015
GROUP BY DEST_COUNTRY_NAME
""")

dataFrameWay = flightData2015 \
    .groupBy("DEST_COUNTRY_NAME") \
    .count()

sqlWay.explain()
dataFrameWay.explain()

  • 실행결과

    image

  • 다음은 멀티 트랜스포메이션 쿼리에 대한 예제이다.

maxSql = spark.sql("""
SELECT DEST_COUNTRY_NAME, sum(count) as destination_total
FROM flight_data_2015
GROUP BY DEST_COUNTRY_NAME
ORDER BY sum(count) DESC
LIMIT 5
""")

maxSql.show()

  • 실행결과

    image

from pyspark.sql.functions import desc

flightData2015 \
    .groupBy("DEST_COUNTRY_NAME") \
    .sum("count") \
    .withColumnRenamed("sum(count)", "destination_total") \
    .sort(desc("destination_total")) \
    .limit(5) \
    .show()

  • 실행결과

    image

  • DataFrame의 explain() 메서드로 확인해보면 이 예제의 물리 실행 계획은 총 7가지 단계로 구성된다.
  • 실행 계획(plan)은 트랜스포메이션의 지향성 비순환 그래프(directed acyclic graph, DAG)이며 액션이 호출되면 결과를 만들어낸다.
    • DAG의 각 단계는 불변성을 가진 신규 DataFrame을 생성한다.

image

  • 스파크 웹 UI의 SQL 탭에서는 실행한 쿼리의 DAG를 그래프로 시각화한다. 6

Spark 웹 UI SQL 탭의 질의 실행 DAG - 부분 집계 HashAggregate가 WholeStageCodegen으로 묶이고, Exchange(셔플)를 지나 최종 HashAggregate와 CollectLimit로 이어진다

출처: Apache Spark 3.3.0 문서 - Web UI (https://archive.apache.org/dist/spark/docs/3.3.0/web-ui.html), Apache License 2.0

flightData2015 \
    .groupBy("DEST_COUNTRY_NAME") \
    .sum("count") \
    .withColumnRenamed("sum(count)", "destination_total") \
    .sort(desc("destination_total")) \
    .limit(5) \
    .explain()

  • 실행결과

    image

1.3 한계와 트레이드오프

  • CSV 파일은 컬럼 타입 정보를 담지 않으므로 inferSchema로 추론해야 하고, 공식 문서 기준으로 이때 데이터를 한 번 더 훑는 추가 패스가 필요하다. 스키마를 미리 알고 있다면 명시적인 스키마를 지정하는 편이 낫다. 1
  • spark.sql.shuffle.partitions의 기본값 200은 데이터 크기와 무관한 고정값이다. 셔플 대상이 작으면 파티션 수만큼 태스크 오버헤드가 늘고, 반대로 데이터가 크면 파티션당 처리할 데이터가 많아져 셔플 단계의 메모리·디스크 부담이 커진다. 4
  • 정형 스트리밍에서는 같은 체크포인트 위치로 쿼리를 재시작할 때 이 값을 바꿀 수 없으므로, 워크로드에 맞는 값을 처음부터 정해두는 편이 좋다. 4
  • DataFrame은 지연 연산으로 계보만 쌓아두므로, 같은 DataFrame을 여러 액션에서 반복해 쓰면 액션마다 계보를 따라 다시 계산될 수 있다. 이럴 때는 캐싱(cache()/persist())으로 재계산을 줄이는 방법을 검토한다.
  • 이 글의 실행 결과는 스파크 3.3.0·로컬 모드·책의 항공운항 요약 데이터(2015-summary.csv) 조건에서 얻은 것이다. 클러스터 구성과 데이터 크기가 다르면 실행 계획의 단계 수와 셔플 파티션 수는 달라질 수 있다.

References

  1. Spark 3.3.0 Documentation - CSV Files (Data Source Option inferSchema) ↩ ↩2

  2. [Spark] 빅데이터와 아파치 스파크란 - ingu627 ↩

  3. PySpark 3.3.0 Documentation - pyspark.sql.DataFrame.explain ↩

  4. Spark 3.3.0 Documentation - Configuration (spark.sql.shuffle.partitions) ↩ ↩2 ↩3

  5. PySpark 3.3.0 Documentation - pyspark.sql.DataFrame.createOrReplaceTempView ↩

  6. Spark 3.3.0 Documentation - Web UI (SQL tab) ↩

댓글남기기

참고