[Spark] SparkSQL 개념 및 코드

 

 

SparkSQL

 

Structured Data를 다루기 위해 Spark에서도 SQL문을 쓸 수 있게 했다.

 

1. Structured Data와 Unstructured Data의 차이점은 무엇이고

2. 왜 RDD가 아닌 Structured Data를 써야 하며

3. Spark에서 SQL문을 어떻게 쓸 수 있을까?

 

포스팅을 읽어가며 질문을 해소해보자.

 

 

 

"본 포스팅은 패스트캠퍼스의 강의를 듣고, 정리한 자료임을 밝힙니다."

 

 

 

 

 

 

Structured Data vs Unstructured Data

---

Unstructured Data : free form

- 로그 파일

- 이미지

말 그대로 free form. 자유로운 형식의 데이터다. 이미지가 될 수도 있고, 하나의 스트링이 될 수 있다. 이런 데이터들을 정제하여 Structured Data로 변환한다. 

 

 

Semi Structured Data : 행과 열

- CSV

- JSON

- XML

기본적인 행과 열은 갖춰져 있는 형태의 데이터다. 하지만 데이터의 타입이 숫자인지, 문자인지 등의 정보는 담고 있는 않는다. 딱 데이터의 존재만 보여주는 형태다. 

 

 

Structured Data : 행과 열 + 스키마

- 데이터 베이스 (ex. table 형태, dataframe)

행과 열을 더불어, 데이터의 타입까지 저장해둔 형태다. 숫자, 문자, 리스트 등 Schema를 담고 있어, 다른 데이터 형태에 비해 최적화에 유리하다.

 

 

 

* RDD가 아닌 Structured Data를 Spark에서 사용하는 이유

 

RDD는 데이터의 구조를 모르기 때문에, 최적화 등 데이터 다루기를 개발자에게 모두 의존해야 한다. 예를 들어, 전 포스팅에서 이야기했던 성능 비교 문제를 떠올려 보자. (1) filter()를 적용한 다음, reduceByKey()를 한 것과 (2) reduceByKey()를 적용한 다음, filter()를 적용한 것은 성능 차이가 존재했다. 이처럼 RDD를 이용하면 개발자의 실력에 따라 성능이 천차만별이 된다. 

하지만 Structured Data의 경우, 구조를 컴퓨터가 알고 있기에 어떤 task를 수행할지만 입력하면 자동으로 최적화까지 진행시킨다. 개발자는 task만 정의 내리면 되는 것이다. 이런 니즈에서 탄생한 것이 SparkSQL이다. 

SparkSQL은 Structured Data를 다룰 수 있게 함으로써, Spark의 이용 성능을 상향 평준화시켰다. 

 

 

 

 

 

 

SparkSQL 간단 설명

---

SparkSQL은 스파크 위에 구현된 하나의 패키지다. 

 

3개의 주요 API가 있는데 SQL, DataFrame, Datasets(pyspark 사용자는 신경 X)이다.

즉, SparkSQL을 사용하면 SQL문으로 데이터를 만질 수 있으며, DataFrame을 형성 및 전처리할 수 있다는 뜻이다.

 

2개의 백엔드 컴포넌트가 존재하는데, Catalyst와 Tungsten이 그것이다. 기능은 다음과 같다. 

Catalyst - 쿼리 최적화 엔진

Tungsten - 시리얼라이저 (용량 최적화)

백엔드에 대한 자세한 내용은 이후 포스팅에서 다루겠다. 

 

 

 

 

 

SparkSQL 사용하기

---

Import

 

import os
import findspark
findspark.init(os.environ.get("SPARK_HOME"))
import pyspark
from pyspark import SparkConf, SparkContext
import pandas as pd
import faulthandler
faulthandler.enable()
from pyspark.sql import SparkSession

 

 

 

SparkCore에선 SparkContext를 이용했다면, SparkSQL에선 SparkSession을 이용한다.

 

# SparkSession을 먼저 입력
# local 환경에서 돌리는 경우, 'local'이라고 설정
# appName()은 자신이 원하는 이름으로 하면 된다.
# SparkSession은 SparkContext와 마찬가지로 1회 불러올 수 있다. 

spark = SparkSession.builder.master('local').appName("sql-learn").getOrCreate()

 

 

 

createDataFrame() 함수를 이용하여 DataFrame을 형성한다.

 

# Data

stocks = [
    ('Google', 'GOOGL', 'USA', 2984, 'USD'), 
    ('Netflix', 'NFLX', 'USA', 645, 'USD'),
    ('Amazon', 'AMZN', 'USA', 3518, 'USD'),
    ('Tesla', 'TSLA', 'USA', 1222, 'USD'),
    ('Tencent', '0700', 'Hong Kong', 483, 'HKD'),
    ('Toyota', '7203', 'Japan', 2006, 'JPY'),
    ('Samsung', '005930', 'Korea', 70600, 'KRW'),
    ('Kakao', '035720', 'Korea', 125000, 'KRW'),
]


# Schema

stockSchema = ['name', 'ticker', 'country', 'price', 'currency']


# createDataFrame()
df = spark.createDataFrame(data = stocks, schema=stockSchema)

 

df.dtypes


[('name', 'string'),
 ('ticker', 'string'),
 ('country', 'string'),
 ('price', 'bigint'),
 ('currency', 'string')]

----

df.show()


+-------+------+---------+------+--------+
|   name|ticker|  country| price|currency|
+-------+------+---------+------+--------+
| Google| GOOGL|      USA|  2984|     USD|
|Netflix|  NFLX|      USA|   645|     USD|
| Amazon|  AMZN|      USA|  3518|     USD|
|  Tesla|  TSLA|      USA|  1222|     USD|
|Tencent|  0700|Hong Kong|   483|     HKD|
| Toyota|  7203|    Japan|  2006|     JPY|
|Samsung|005930|    Korea| 70600|     KRW|
|  Kakao|035720|    Korea|125000|     KRW|
+-------+------+---------+------+--------+

 

 

 

 

createOrReplaceTempView() 이용

이 함수를 통해 DataFrame을 SQL문으로 불러오고, 전처리할 수 있다. 

 

# dataframe을 먼저 입력
# 함수 입력
# createOrReplaceTempView() 함수 안쪽에는 SQL문 쓸 때 데이터프레임을 어떤 이름으로 사용할 것인지
# 정의한다. 

df.createOrReplaceTempView('stocks')

 

 

 

SQL문으로 자유롭게 데이터를 다뤄보자.

 

spark.sql("select name from stocks").show()


+--------------+
|(price > 2000)|
+--------------+
|          true|
|         false|
|          true|
|         false|
|         false|
|          true|
|          true|
|          true|
+--------------+


----


spark.sql("select name, price from stocks where country = 'Korea'").show()


+-------+------+
|   name| price|
+-------+------+
|Samsung| 70600|
|  Kakao|125000|
+-------+------+


----


spark.sql("select name, price from stocks order by price asc").show()


+-------+------+
|   name| price|
+-------+------+
|Tencent|   483|
|Netflix|   645|
|  Tesla|  1222|
| Toyota|  2006|
| Google|  2984|
| Amazon|  3518|
|Samsung| 70600|
|  Kakao|125000|
+-------+------+


----


spark.sql("select count(price) from stocks where country in ('Korea', 'USA')").show()


+------------+
|count(price)|
+------------+
|           6|
+------------+

 

 

 

 

다른 데이터를 Join 해보자.

 

earnings = [
    ('Google', 27.99, 'USD'), 
    ('Netflix', 2.56, 'USD'),
    ('Amazon', 6.12, 'USD'),
    ('Tesla', 1.86, 'USD'),
    ('Tencent', 11.01, 'HKD'),
    ('Toyota', 224.82, 'JPY'),
    ('Samsung', 1780., 'KRW'),
    ('Kakao', 705., 'KRW')
]


from pyspark.sql.types import StringType, FloatType, StructType, StructField
# Schema에서 DataType을 직접 지정해줄 수 있다. 

earningsSchema = StructType([
    StructField("name", StringType(), True),
    StructField("EPS", StringType(), True),
    StructField("currency", StringType(), True)
])


df_earning = spark.createDataFrame(data=earnings, schema=earningsSchema)
df_earning.createOrReplaceTempView('earnings')

 

 

 

SQL문을 쓰지 않고도, 데이터를 다룰 순 있다. 하지만 SQL문을 통해 더 많이 다룬다.

 

df_earning.select("*").show()


+-------+------+--------+
|   name|   EPS|currency|
+-------+------+--------+
| Google| 27.99|     USD|
|Netflix|  2.56|     USD|
| Amazon|  6.12|     USD|
|  Tesla|  1.86|     USD|
|Tencent| 11.01|     HKD|
| Toyota|224.82|     JPY|
|Samsung|1780.0|     KRW|
|  Kakao| 705.0|     KRW|
+-------+------+--------+

 

 

 

JOIN 하여 데이터를 더 다양한 각도에서 살펴보자!

 

spark.sql("select * from stocks join earnings on stocks.name = earnings.name").show()


+-------+------+---------+------+--------+-------+------+--------+
|   name|ticker|  country| price|currency|   name|   EPS|currency|
+-------+------+---------+------+--------+-------+------+--------+
| Amazon|  AMZN|      USA|  3518|     USD| Amazon|  6.12|     USD|
| Google| GOOGL|      USA|  2984|     USD| Google| 27.99|     USD|
|  Kakao|035720|    Korea|125000|     KRW|  Kakao| 705.0|     KRW|
|Netflix|  NFLX|      USA|   645|     USD|Netflix|  2.56|     USD|
|Samsung|005930|    Korea| 70600|     KRW|Samsung|1780.0|     KRW|
|Tencent|  0700|Hong Kong|   483|     HKD|Tencent| 11.01|     HKD|
|  Tesla|  TSLA|      USA|  1222|     USD|  Tesla|  1.86|     USD|
| Toyota|  7203|    Japan|  2006|     JPY| Toyota|224.82|     JPY|
+-------+------+---------+------+--------+-------+------+--------+


----

# PER = Price / EPS
spark.sql("select stocks.name, (stocks.price/earnings.EPS) AS PER from stocks join earnings on stocks.name = earnings.name").show()


+-------+------------------+
|   name|               PER|
+-------+------------------+
| Amazon| 574.8366013071895|
| Google|106.60950339406932|
|  Kakao| 177.3049645390071|
|Netflix|        251.953125|
|Samsung|39.662921348314605|
|Tencent|43.869209809264305|
|  Tesla|  656.989247311828|
| Toyota| 8.922693710523975|
+-------+------------------+

 

 

 

 

지금까지 SparkSQL을 사용하는 방법을 간략하게 알아보았다.

다음 포스팅에선 DataFrame을 더 자세하게 다뤄보도록 하겠다.

수고했습니다!