📒 Spark(9)

📌 Spark 파일포맷

• Unstructured

  • Text

• Semi-structured

  • JSON
  • XML
  • CSV

• Structured

  • PARQUET
  • AVRO
  • ORC
  • SquenceFile

특징	CSV	JSON	PARQUET	AVRO
Column Storage	X	X	Y	X
압축 기능	Y	Y	Y	Y
Splittable	Y	Y	Y	Y
Human readable	Y	Y	X	X
Nested Structure support	X	Y	Y	Y
Schema evolution	X	X	Y	Y

• CSV, JSON : 압축되면 Splittable하지 않다.
  -> 압축 방식에 따라 다르다. - snappy 압축이라면 Splittable)
• PARQUET : Spark의 기본 파일 포맷
• gzip으로 압축된 CSV, JSON 파일은 Splittable하지 않기 때문에 하나의 executor가 일단 처리하게 되며 메모리 에러가 생길 확률이 높다.

Parquet : Spark의 기본 파일 포맷

• Twiter & Cloudera 에서 공동 개발하였다. (Doug Cutting)
  

• Hybrid Storage (Row Group)
  -> Paruet가 사용하는 방식이다. 하나의 데이터 블록은 하나의 Row Group으로 구성되었다.

📌 Execution Plan

Transformations & Actions

• Transformations

  • Narrow Dependencies : 독립적인 Partition level작업
    -> SELECT, FILTER, MAP 등등
  • Wide Dependencies: Shuffling이 필요한 작업
    -> GROUP BY, REDUCE BY, PARTITION BY, REPARTITION 등등

• Actions

  • Read, Write, Show, Collect -> Job을 실행 시킨다. -> 실제 코드가 실행된다.
  • Lazy Execution
    -> 더 많은 오퍼레이션을 볼 수 있기에 최적화를 더 잘할 수 있다.
    -> 때문에 SQL이 더 선호된다.

Transformations & Actions Visualization

spark.read.option("header", True).csv("test.csv"). \
	where("gender <> 'F'"). \
    select("name", "gender"). \
    groupby("gender). \
    count(). \
    show()

Jobs, Stages, Tasks

• Action -> Job -> 1+Stages -> 1+Tasks

• Action
  -> Job을 하나 만들어내고 코드가 실제로 실행된다.

• Job

  • 하나 혹은 그 이상의 Stage로 구성되었다.
  • Stage는 Shuffling이 발생하는 경우 새로 생긴다.

• Stage

  • DAG의 형태로 구성된 Task들이 존재한다.
  • 여기 Task들은 병렬 실행이 가능하다.

• Task
  -> 가장 작은 실행 유닛으로 Executor에 의해 실행된다.

📌 Bucketing & Partitioning

Bucketing & Partitioning 소개

• 둘다 Hive 메타스토어의 사용이 필효하다 -> saveAsTable

• 데이터 저장을 이후 반복처리에 최적화된 방법으로 하는 것이다.

• Bucketing

  • 먼저 Aggregation이나 Window 함수나 Join에서 많이 사용되는 컬럼이 있는가?
  • 있다면 데이터를 이 특적 컬럼(들)을 기준으로 테이블로 저장.
    -> 이 때의 버킷의 수도 지정하다.

• FileSystem Partitioning

  • 원래 Hive에서 많이 사용한다.
  • 데이터의 특정 컬럼(들)을 기준으로 폴더 구조를 만들어 데이터 저장을 최적화한다.
    -> 위의 컬럼들을 Partition Key 라고 부른다.

Buketing

• DataFrame을 특정 ID를 기준으로 나누어서 테이블로 저장한다.

  • 다음부터는 이를 로딩하여 사용함으로써 반복 처리시 시간을 단축할 수 있다.

    • DataFrameWriter의 buketBy 함수를 사용한다.
      -> Bucket의 수와 기준 ID를 지정한다.

  • 데이터의 특성을 잘 알고 있는 경우 사용 가능하다.

File System Partitioning

• 데이터를 Partition Key 기반 폴더 ("Partition") 구조로 물리적으로 나누어서 저장한다.
  -> Hive에서 사용하는 Partitioning을 말한다.

• Partitioning 예

  • 큰 로그 파일을 데이터 생성시간을 기반으로 데이터읽기를 많이 한다면?
    -> 데이터 자체를 연-월-일의 구조로 저장한다.
    -> 보통 이와같이 이미 저장되어 있는 경우가 많다.

• Partitioning의 장점

  • 데이터의 읽기 과정을 최적화할 수 있다.(Scaning 과정이 줄어들거나 없어진다.)
  • 데이터 관리도 쉬워진다. (Retention Policy를 적용할 경우)

• DataFrameWriter의 paritionBy를 사용한다.
  -> Partition key를 잘못 선택하면 엄청나게 많은 파일들이 생성된다.
  -> Partition key는 카디널리티가 낮은것을 사용해야한다.