아파치 스파크 SparkSession

SparkSession는 Apache Spark 애플리케이션의 진입점(Entry Point)으로, 스파크 SQL, 데이터프레임, 그리고 Dataset API를 포함한 다양한 기능에 접근할 수 있도록 하는 핵심 객체이다. SparkSession은 이전 버전의 SQLContext, HiveContext, 그리고 SparkContext의 기능을 통합하여 제공함으로써, 사용자가 보다 간편하게 Spark 애플리케이션을 구성하고 실행할 수 있도록 한다.

1 개요

SparkSession은 Apache Spark 2.0부터 도입되었으며, 분산 데이터 처리, SQL 쿼리, 데이터 분석 및 머신러닝 작업을 수행하는 데 필요한 환경을 구성한다. 이를 통해 데이터 소스(CSV, JSON, Parquet, ORC, Hive 등)를 손쉽게 읽고, 데이터프레임(DataFrame)이나 Dataset을 생성하여 처리할 수 있다.

2 주요 기능

• 단일 진입점 (Unified Entry Point)
  • SparkSession은 SparkContext, SQLContext, HiveContext 등 여러 컨텍스트를 통합하여, 사용자가 하나의 객체를 통해 다양한 API에 접근할 수 있도록 한다.
• 데이터 소스 통합 (Integrated Data Source Support)
  • 다양한 파일 형식 및 외부 데이터베이스와의 연결을 지원하며, 데이터 읽기와 저장 작업을 단순화한다.
• Catalyst 옵티마이저 및 Tungsten 실행 엔진 통합
  • 데이터프레임 연산에 대해 자동으로 최적화된 실행 계획을 수립하여, 고성능 분산 데이터 처리를 가능하게 한다.
• Spark SQL 지원
  • SQL 쿼리를 통해 데이터프레임에 저장된 데이터를 조작하고 분석할 수 있으며, 임시 뷰를 생성하여 복잡한 쿼리도 수행할 수 있다.

3 사용 방법

SparkSession을 사용하여 Spark 애플리케이션을 구성하는 기본 단계는 다음과 같다.

• SparkSession 생성
  • SparkSession.builder를 사용하여 애플리케이션 이름, 마스터 URL, 구성 옵션 등을 지정한 후 getOrCreate()를 호출한다.
• 데이터 소스에서 데이터 읽기
  • spark.read를 통해 CSV, JSON, Parquet 등 다양한 형식의 데이터를 데이터프레임으로 불러온다.
• 데이터 처리 및 SQL 쿼리 실행
  • 데이터프레임 API를 사용하여 데이터를 변환하고, createOrReplaceTempView를 통해 임시 뷰를 생성한 후 SQL 쿼리를 실행한다.
• 애플리케이션 종료
  • 작업이 완료되면 spark.stop()을 호출하여 리소스를 정리한다.

4 예제 코드 (Python)

다음은 pyspark를 사용하여 SparkSession을 생성하고 데이터프레임을 처리하는 예제이다.

from pyspark.sql import SparkSession

# SparkSession 생성: 애플리케이션 이름 및 설정 지정
spark = SparkSession.builder \
    .appName("SparkSessionExample") \
    .master("local[*]") \
    .getOrCreate()

# CSV 파일에서 데이터프레임 생성 (헤더 포함, 스키마 자동 추론)
df = spark.read.csv("data.csv", header=True, inferSchema=True)

# 데이터프레임 스키마 출력
df.printSchema()

# 데이터프레임의 일부 행 조회
df.show(5)

# 임시 뷰 생성 및 SQL 쿼리 실행
df.createOrReplaceTempView("data_table")
result = spark.sql("SELECT column1, COUNT(*) AS count FROM data_table GROUP BY column1")
result.show()

# SparkSession 종료
spark.stop()

5 예제 코드 (Scala)

다음은 스칼라(Scala)를 사용하여 SparkSession을 생성하고 데이터프레임을 처리하는 예제이다.

import org.apache.spark.sql.SparkSession

object SparkSessionExample {
  def main(args: Array[String]): Unit = {
    // SparkSession 생성 (로컬 모드 실행)
    val spark = SparkSession.builder
      .appName("SparkSessionExample")
      .master("local[*]")
      .getOrCreate()

    // CSV 파일에서 데이터프레임 생성 (헤더 옵션과 스키마 추론)
    val df = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("data.csv")

    // 데이터프레임 스키마 출력
    df.printSchema()

    // 데이터프레임의 일부 행 조회
    df.show(5)

    // 임시 뷰 생성 및 SQL 쿼리 실행
    df.createOrReplaceTempView("data_table")
    val result = spark.sql("SELECT column1, COUNT(*) AS count FROM data_table GROUP BY column1")
    result.show()

    // SparkSession 종료
    spark.stop()
  }
}

6 역사 및 배경

SparkSession은 Apache Spark 2.0 버전에서 도입되었으며, 이전 버전에서는 SparkContext, SQLContext, HiveContext 등 여러 개의 객체를 사용하여 환경을 구성하였다. 이들 개별 컨텍스트를 하나의 통합된 인터페이스로 제공함으로써, 개발자는 보다 간편하게 애플리케이션을 개발할 수 있게 되었고, 동시에 Catalyst 옵티마이저와 Tungsten 실행 엔진의 최적화 기능을 활용하여 고성능 데이터 처리를 실현할 수 있게 되었다.

7 활용

SparkSession은 데이터 분석, ETL(Extract, Transform, Load) 작업, 머신러닝, 실시간 스트리밍 등 다양한 분야에서 활용된다.

• 다양한 데이터 소스와의 연동을 통해 대규모 데이터를 손쉽게 처리할 수 있다.
• Spark SQL 및 데이터프레임 API를 활용하여, SQL 쿼리와 유사한 방식으로 데이터를 조작하고 분석할 수 있다.
• Catalyst 옵티마이저와 Tungsten 실행 엔진의 통합으로, 복잡한 데이터 변환과 분석 작업을 최적화된 방식으로 수행할 수 있다.

8 같이 보기

• 아파치 스파크
• RDD (Resilient Distributed Dataset)
• 스파크 SQL
• Catalyst 옵티마이저
• Tungsten (스파크 실행 엔진)

9 참고 문헌

• Zaharia, M., Chowdhury, M., Franklin, M. J., Shenker, S., & Stoica, I. (2010). Spark: Cluster Computing with Working Sets. HotCloud.
• Apache Spark 공식 문서, https://spark.apache.org/docs/latest/sql-programming-guide.html
• Karau, H., Konwinski, A., Wendell, P., & Zaharia, M. (2015). Learning Spark: Lightning-Fast Data Analytics. O'Reilly Media.