[스파크 완벽 가이드] 10장 - 스파크 SQL

스파크 SQL 쿼리 실행 방법

1. 스파크 SQL CLI
2. 스파크의 프로그래밍 SQL 인터페이스
3. 스파크 SQL Thrift JDBC/ODBC 서버

 

카탈로그

스파크 SQL에서 가장 높은 추상화 단계는 카탈로그이다.

카탈로그는 테이블에 저장된 데이터에 대한 메타데이터뿐만 아니라 데이터베이스, 테이블, 함수 그리고 뷰에 대한 정보를 추상화한다.

 

테이블

테이블은 명령을 실행할 데이터의 구조라는 점에서 DataFrame과 논리적으로 동일하다.

스파크에서 테이블을 생성하면 default 데이터베이스에 등록된다.

 

테이블은 항상 데이터를 가지고 있다.

임시 테이블의 개념이 없으며 데이터를 가지지 않는 뷰만 존재한다.

따라서 테이블을 제거하면 모든 데이터가 사라지므로 주의해야 한다.

 

<스파크 관리형 테이블>

관리형 테이블은 스파크가 데이터 파일을 완전히 관리하는 테이블 유형이다.

테이블을 생성할 때 스파크가 데이터의 저장 위치와 메타데이터를 모두 관리하며, 메타스토어에서 해당 테이블을 추적한다.

 

외부 테이블은 데이터의 물리적 위치가 스파크 또는 하이브 외부에서 관리되는 테이블이다.

스파크는 메타데이터만 관리하고, 데이터 파일은 사용자가 관리한다.

 

<테이블 생성하기>

스파크는 다양한 데이터소스를 사용해 테이블을 생성할 수 있고, 다음과 같은 명령어를 사용한다.

CREATE TABLE flights (
  DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count LONG)
USING JSON OPTIONS (path '경로')

USING 구문을 사용하여 포맷을 지정하지 않으면, 스파크는 하이브 SerDe 설정을 사용한다.

하이브 SerDe는 스파크의 자체 직렬화보다 훨씬 느리므로 성능에 영향을 미친다.

 

<파티셔닝된 테이블에 데이터 삽입>

INSERT INTO partitioned_flights
  PARTITION (DEST_COUNTRY_NAME = "UNITED STATES")
  SELECT count, ORIGIN_COUNTRY_NAME
  FROM flights
  WHERE DEST_COUNTRY_NAME = "UNITED STATES"
  LIMIT 12

 

파티셔닝된 테이블 partitioned_flight에 flights 테이블의 일부 데이터를 삽입한다.

 

뷰

뷰는 실질적으로 트랜스포메이션이며 스파크는 쿼리가 실행될 때만 뷰를 실행한다.

• 임시 뷰 - 데이터베이스에 등록되지 않고 현재 세션에서만 사용
• 전역 임시 뷰 - 전체 스파크 애플리케이션에서 볼 수 있는 뷰

 

복합 데이터 타입 쿼리

스파크 SQL에는 구조체, 리스트, 맵 세 가지 복합 데이터 타입이 존재한다.

 

<구조체>

구조체는 맵에 더 가까우며 스파크에서 중첩 데이터를 생성하거나 쿼리하는 방법을 제공한다.

CREATE VIEW IF NOT EXISTS nested_data AS
  SELECT (DEST_COUNTRY_NAME, ORIGIN_COUNTRY_NAME) country, count
  FROM flights

위와 같이 여러 컬럼이나 표현식을 괄호로 묶기만 하면 생성할 수 있다.

 

SELECT country.DEST_COUNTRY_NAME, count
FROM nested_data

구조체의 개별 컬럼을 조회할 수 있다.

 

<리스트>

collect_list 함수나 중복 없는 배열을 만드는 collect_set 함수를 사용할 수 있다.

SELECT DEST_COUNTRY_NAME new_name, collect_list(count) flight_counts,
  collect_set(ORIGIN_COUNTRY_NAME) origin_set
FROM flights
GROUP BY DEST_COUNTRY_NAME

 

직접 배열을 생성할 수 있다.

SELECT DEST_COUNTRY_NAME, ARRAY(1, 2, 3)
FROM flights

 

파이썬과 유사한 방법으로 특정 위치의 데이터를 쿼리할 수 있다.

SELECT DEST_COUNTRY_NAME new_name, collect_list(count)[0]
FROM flights
GROUP BY DEST_COUNTRY_NAME

 

explode 함수를 사용해 배열을 다시 여러 로우로 변환할 수 있다.

CREATE OR REPLACE TEMP VIEW flights_agg AS
  SELECT DEST_COUNTRY_NAME, collect_list(count) collected_counts
  FROM flights
  GROUP BY DEST_COUNTRY_NAME
 
-- explode 사용
SELECT explode(collected_counts), DEST_COUNTRY_NAME
FROM flights_add

 

사용자 정의 함수

def power3(number:Double): Double = number * number * number
spark.udf.register("power3", power3(_:Double):Double)

SELECT count, power3(count)
FROM flights