[ Spark ] 스파크 간단 스터디 5

Chapter 11. Dataset¶

• 구조적 api 의 기본 데이터 타입
• 데이터프레임은 row 타입의 dataset 이다.
• 자바 가상 머신을 사용하는 언어인 스칼라 와 자바 에서만 사용할 수 있다.
• 타입형 api 라고 불린다
• 데이터프레임 api 를 사용할 때 string, integer 데이터 타입의 객체를 생성하지 않지만 스파크는 row 객체를 변환해 데이터를 처리
• 스칼라에서 데이터프레임은 row 타입의 dataset 을 의미함
• 인코더 : 도메인별 특정 객체 T를 스파크의 내부 데이터 타입으로 매핑하는 시스템
• DatasetAPI 를 사용하면 스파크는 데이터셋에 접근할 대마다 row 포맷이 아닌 사용자 정의 데이터 타입으로 변환
• 이 변환 작업은 느리지만 사용자에게 더 많은 유연성을 제공함

인코더의 역할¶

• 런타임 환경에서 객체를 바이너리 구조로 직렬화하는 코드를 생성하도록 스파크에 지시
• row 타입을 직렬화된 바이너리 구조로 변환

11.1 Dataset 을 사용할 시기¶

• Dataset 을 사용해야하는 이유

  DataFrame 기능만으로는 수행할 연산을 표현할 수 없는 경우,

  성능 저하를 감수하더라도 타입안정성을 가진 데이터 타입을 사용하고 싶은 경우

• 복잡한 비즈니스 로직을 단일함수로 인코딩해야하는 경우 > 이런경우 Dataset API 사용이 유리(타입안정성 덕분에)

• 정확도와 방어적 코드를 중요시하는 경우 > 성능을 조금 희생하더라도 Dataset API 가 유리
• 잘못된 데이터로부터 애플리케이션 보호할 순 없지만, 우아하게 데이터를 제어, 구조화 가능
• 단일노드의 워크로드와 스파크 워크로드에서 전체 로우에 대한 다양한 트랜스포메이션을 DatasetAPI 에 사용하는 것이 적합
• 장점중 하나는 로컬과 분산환경의 워크로드 재사용 가능한 것이 DF 의 장점

11.2 dataset 생성 - 스파크¶

• 스칼라 case class 구문을 사용해 데이터 타입을 정한다
• case class 구문의 특징 : (불병성,패턴 매칭으로 분해 기능,참조값 대신 클래스르이다,사용하기 쉽고 다루기 판함)

In [1]:

case class Flight(DEST_COUNTRY_NAME : String,
                  ORIGIN_COUNTRY_NAME: String, count: BigInt)

Intitializing Scala interpreter ...

Spark Web UI available at http://win10-210321GO:4040
SparkContext available as 'sc' (version = 2.4.8, master = local[*], app id = local-1625999362845)
SparkSession available as 'spark'

Out[1]:

defined class Flight

In [2]:

val flightsDF = spark.read.parquet("C:/Users/COM/Desktop/home/study_group/1st_spark_studygroup_2106_/Spark-The-Definitive-Guide-master/data/flight-data/parquet/part-00000-f12e2c87-b1bd-4d31-b8f9-769d78936722-c000.snappy.parquet")
val flights = flightsDF.as[Flight]

Out[2]:

flightsDF: org.apache.spark.sql.DataFrame = [DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string ... 1 more field]
flights: org.apache.spark.sql.Dataset[Flight] = [DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string ... 1 more field]

• Flight 데이터 타입의 Dataset 을 생성
• Flight 데이터 타입은 스키마만 정의되어 있을 뿐 아무런 메서드도 정의되어 있지 않음
• 데이터를 읽으면 DataFrame 이 반환

11.3 액션¶

• collect, take, count 같은 액션을 적용할 수 있다는 사실

In [3]:

flights.show(2)

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|    United States|            Romania|    1|
|    United States|            Ireland|  264|
+-----------------+-------------------+-----+
only showing top 2 rows

• 케이스 클래스에 접근할 때 어떠한 데이터 타입도 필요하지 않다는 사실
• 케이스 클래스의 속성명을 지정하면 속성에 맞는 값과 데이터 타입 모두 반환

In [4]:

flights.first.DEST_COUNTRY_NAME

Out[4]:

res1: String = United States

11.4 트랜스포메이션¶

• 데이터셋의 트랜스포메이션은 데이터프레임과 동일하며 데이터 프레임의 모든 트랜스포메이션은 데이터셋에서 사용 가능
• 데이터셋은 원형의 JVM 데이터 타입을 다루기에 복잡하고 강력한 데이터 타입으로 트랜스포메이션을 사용 가능

• 아래는 불리언 값을 반환하는 함수
• 불리언 값은 출발지와 도착지가 동일한지 나타냄

In [5]:

def originIsDestination(flight_row: Flight): Boolean = {
    return flight_row.ORIGIN_COUNTRY_NAME == flight_row.DEST_COUNTRY_NAME
}

Out[5]:

originIsDestination: (flight_row: Flight)Boolean

• 위에서 정의한 함수를 fliter 메서드에 적용해 각 행이 true를 반환하는지 평가하고 데이터셋을 필터링 가능

• 필터링은 단순한 트랜스포메이션이며 특정값을 다른값으로 매핑하기도 함
• 데이터 프레임에서 매핑 작업을 수행하는 것은 데이터셋의 select 메서드를 사용하는 것과 같음

In [7]:

val destination = flights.map(f => f.DEST_COUNTRY_NAME)

Out[7]:

destination: org.apache.spark.sql.Dataset[String] = [value: string]

• 스파크는 결과로 반환할 JVM 데이터 타입을 알고 있기 때문에 컴파일시에 데이터 타입의 유효성 검사 가능
• 데이터 프레임을 사용하면 코드 생성 기능과 같은 장점이 있음
• 하지만, 매핑 작업을 사용한다면 훨씬 정교하게 로우 단위로 처리 가능

11.5 조인¶

• 조인은 데이터프레임과 마찬가지로 데이터 셋에도 동일하게 적용
• 데이터 셋은 joinWith 처럼 정교한 메서드 제공 : co-group 과 거의 유사하며 데이터셋 안족에 다른 두개의 중첩된 데이터 셋으로 구성
• 각 컬럼은 단일 데이터셋이므로 조인 수행 시 더 많은 정보를 유지할 수 있으며 고급 맵이나 필터처럼 정교하게 데이터를 다룰 수 있음

11.6 그룹화 집계¶

• groupBy, rollup, cube 메서드를 여전히 사ㅣ용 가능
• 하지만, 데이터셋 대신 데이터 프레임을 반환하기 때문에 데이터 타입 정보를 잃음
• 정보를 잃는 것은 큰 문제는 아니지만, 이를 유지할 수 있는 그룹화와 집계 방법 존재
• groupByKey 메서드는 데이터셋의 특정 키를 기준으로 그룹화하고 형식화된 데이터셋을 반환

In [ ]: