📒 Spark(5)

📌 Spark Session 생성

• Spark 프로그램의 시작은 SparkSession을 만드는 것이다.

  • 프로그램마다 하나를 만들어서 Spakr Cluster와 통신한다. -> Singleton 객체
  • Spark 2.0에서 처음 소개되었다.

• Spark Session을 통해 Spark이 제공해주는 다양한 기능을 사용한다.

  • DataFrame, SQL, Streaming, ML API 모두 이 객체로 통신한다.
  • config 메소드를 이용하여 다양한 환경설정이 가능하다.
  • RDD와 관련된 작업을 할때는 SparkSession 아래에 sparkContext 객체를 사용한다.

• Spark Session API

---

📌 Spark Session Create - PySpark 예제

from pyspark.sql import SparkSession
# SparkSession은 Singleton이다.
spark = SparkSession.builder.master("local[*]").appName('PySpark Tutorial').getOrCreate()

spark.stop()

• Spark SQL Engine이 중심으로 돌아간다.

• Singleton

---

📌 pyspark.sql 제공 주요 기능

pyspark.sql.SparkSession

• Spark APP의 엔트리 포인트
• Spark 컨텍스트와 SQL 컨텍스트를 모두 포함하고 있다.
• DataFrame, RDD등을 생성할 수 있는 메서드를 제공한다.

pyspark.sql.DataFrame

• Spark의 분산 데이터 컬렉션
• 구조화된 데이터(행+열)을 저장하고 처리한다.
• SQL 쿼리, 데이터 변환, 필터링 등의 작업 수행이 가능하다.

pyspark.sql.Column

• DataFrame의 열을 나타내는 객체
• 산술 연산, 조건문, 함수 적용 등 다양한 작업 수행이 가능하다.

pyspark.sql.Row

• DataFrame의 행을 나타내는 객체이다.
• 행 데이터에 접근하고 조작이 가능하다.

pyspark.sql.functions

SparkSQL에서 사용 가능한 다양한 함수를 제공한다.

• 문자열 처리, 날짜 처리, 수학 함수 등을 포함한다.

pyspark.sql.types

SparkSQL에서 사용되는 데이터 타입을 정의한다.

pyspark.sql.Window

• 윈도우 함수를 사용하기 위한 객체이다.
• 행 기반 계산을 수행할 때 사용한다. (ex: 순위, 누적합등등)

---

📌 Spark Session 환경변수

• 참조

• executor 별 메모리 : spark.executor.memory (기본 : 1gb)

• executor 별 CPU개수 : spark.executor.cores (YARN에서는 기본값 1)

• driver 메모리 : spark.driver.memory (기본 : 1gb)

• Shuffle 후 Partition의 수 : spark.sql.shuffle.partitions (기본 : 최대 200)

• 사용하는 Resource Manager에 따라 환경변수가 많이 달라진다.

---

📌 Spark Session 환경설정 방법

• 환경변수

• $SPARK_HOME/conf/spark_defaults.conf

• 2개는 보통 Spark Cluster Admin이 관리한다.

• spark-submit 명령의 커맨드라인 파라미터

• SparkSession을 만들때 지정
  -> SparkConf

만약 환경 설정이 충돌될 경우 SparkConf -> Spark-submit -> $SPARK_HOME -> 환경변수 순서의 우선 순위를 가진다.

1. Spark Session 환경 설정 - SparkSession 생성시 일일히 지정

• 이 시점의 Spark Configuration은 앞서 언급한 환경변수와 spark_defaults.conf와 spark-submit로 들어온 환경설정이 우선순위를 고려한 상태로 정리된 상태

from pyspark.sql import SparkSession

spark = SparkSession.builder\
	.master("local[*]")\
	.appName('PySpark Tutorial')\
	.config("spark.some.config.option1", "some-value") \
	.config("spark.some.config.option2", "some-value") \
	.getOrCreate()

2. Spark Session 환경 설정 - SparkConf 객체에 환경 설정하고 SparkSession에 지정

from pyspark.sql import SparkSession
from pyspark import SparkConf

# SparkConf 객체 생성
conf = SparkConf()
# set 메소드를 사용하여 지정
conf.set("spark.app.name", "PySpark Tutorial")
conf.set("spark.master", "local[*]")

spark = SparkSession.builder\
	.config(conf=conf) \
	.getOrCreate()

---

📌 전체적인 흐름

• Spark 세션을 만들기
• 입력 데이터 로딩
• 데이터 조작 작업
  • DataFrame API나 Spark SQL을 사용한다.
  • 원하는 결과가 나올때까지 새로운 DataFrame을 생성한다.
• 최종 결과 저장
  

---

📌 Spark Session이 지원하는 데이터 소스

• 참고

• spark.read(DataFrameReader)를 사용하여 DataFrame으로 로드한다.

• DataFrame.write(DataFrameWriter)을 사용하여 DataFrame을 저장한다.

• 많이 사용되는 데이터 소스

  • HDFS File
    • CSV, JSON, Parquet, ORC, Text, Avro
      - Hive Table
  • JDBC 관계형 DB
  • Cloud 기반 데이터 시스템
  • 스트리밍 시스템