[SparkSQL] Catalyst, Tungsten 작동 원리

 

 

Spark Backend

: Catalyst, Tungsten

 

Spark는 쿼리를 돌리기 위해 위의 두 가지 엔진을 사용한다.

Catalyst는 쿼리문을 최적화 시키는데 이용하고, Tungsten은 RDD level에서 용량을 최적화시킨다.

Process를 보면 다음과 같다.

 

 

 

SparkSQL 백엔드

 

 

 

 

 

Spark의 조직도는 다음과 같다.

 

 

 

Spark 조직도

 

 

 

조직도를 보면 알 수 있다시피 Catalyst는 SQL, DataFrame이 Structured Data를 다룰 수 있게 하는 필수 모듈이다.

그렇다면 Catalyst가 구체적으로 어떤 기능을 하는지 살펴보자.

 

 

 

 

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

 

 

 

 

Catalyst의 기능

---

Logical Plan -> Physical Plan

 

Logical Plan이란? 수행하는 모든 Transformation 단계를 추상화한 것이다. 데이터가 어떻게 변해야 하는지 정의하며, 실제 어디서 어떻게 동작하는지는 정의하지 않는다.

 

Physical Plan이란? Logical Plan이 어떻게 클러스터 위에서 실행될지 정의한다. 실행 전략을 만들고, Cost Model을 통해 후보 Plan들을 평가한다. 가장 성능이 좋은 Plan을 뽑는 식으로 최적화한다.

 

 

 

진행 단계

 

1단계, 분석: DataFrame 객체의 relation을 계산한다. 칼럼의 타입과 이름을 확인해둔다.

2단계, Logical Plan 최적화: (1) 상수로 표현된 식을 Compile Time에 계산 (2) Predicate Pushdown: join & filter -> filter & join (3) Projection Prunning: 연산에 필요한 칼럼만 가져오기

3단계, Physical Plan 만들기: Spark에서 실행 가능한 Plan으로 변환한다.

4단계, Code generation: 최적화된 Physical Plan을 Java Bytecode로 변환.

 

 

 

 

 

 

 

Catalyst의 기능을 코드로 직접 확인해보자.

 

 

Basic Setting

 

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

spark = SparkSession.builder.master('local').appName("dataframe").getOrCreate()

# 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.createOrReplaceTempView('stocks')


# Data 2

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')

 

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|
+-------+------+---------+------+--------+-------+------+--------+

 

 

 

 

 

 

Logical Plan 최적화

다음 코드를 통해 logical plan 최적화를 어떻게 하는지 살펴보자.

 

query = '''
SELECT
    earnings.currency,
    count(*) AS count
FROM
    stocks 
    LEFT JOIN 
        earnings 
    ON 
        stocks.name = earnings.name
    WHERE
        earnings.EPS > 10
    GROUP BY
        earnings.currency
'''

spark.sql(query).show()


+--------+-----+
|currency|count|
+--------+-----+
|     KRW|    2|
|     JPY|    1|
|     HKD|    1|
|     USD|    1|
+--------+-----+

 

 

본래의 Plan은 다음과 같다.

(1) 코드 스캔 -> (2) Join -> (3) Filter -> (4) Project -> (5) Aggregate

 

위의 Plan을 Catalyst가 다음과 같이 최적화한다.

(1) 코드 스캔 -> (2) Filter -> (3) Join -> (4) Project -> (5) Aggregate 

 

 

 

 

 

 

Explain() 함수를 통해 Cataylst의 역할을 세밀하게 살피자.

 

spark.sql(query).explain(True)



# 스캔한 Logical Plan이다. ------------------------------------------------------

== Parsed Logical Plan ==
'Aggregate ['earnings.currency], ['earnings.currency, 'count(1) AS count#875]
+- 'Filter ('earnings.EPS > 10)
   +- 'Join LeftOuter, ('stocks.name = 'earnings.name)
      :- 'UnresolvedRelation [stocks], [], false
      +- 'UnresolvedRelation [earnings], [], false

 

# Logical Plan을 분석한다. ------------------------------------------------------

== Analyzed Logical Plan ==
currency: string, count: bigint
Aggregate [currency#655], [currency#655, count(1) AS count#875L]
+- Filter (cast(EPS#654 as int) > 10)
   +- Join LeftOuter, (name#0 = name#653)
      :- SubqueryAlias stocks
      :  +- View (`stocks`, [name#0,ticker#1,country#2,price#3L,currency#4])
      :     +- LogicalRDD [name#0, ticker#1, country#2, price#3L, currency#4], false
      +- SubqueryAlias earnings
         +- View (`earnings`, [name#653,EPS#654,currency#655])
            +- LogicalRDD [name#653, EPS#654, currency#655], false

 

# Logical Plan을 최적화한다. --------------------------------------------------------------

== Optimized Logical Plan ==
Aggregate [currency#655], [currency#655, count(1) AS count#875L]
+- Project [currency#655]
   +- Join Inner, (name#0 = name#653)
      :- Project [name#0]
      :  +- Filter isnotnull(name#0)
      :     +- LogicalRDD [name#0, ticker#1, country#2, price#3L, currency#4], false
      +- Project [name#653, currency#655]
         +- Filter ((isnotnull(EPS#654) AND (cast(EPS#654 as int) > 10)) AND isnotnull(name#653))
            +- LogicalRDD [name#653, EPS#654, currency#655], false

 

# Physical Plan으로 변환 -----------------------------------------------------------------

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[currency#655], functions=[count(1)], output=[currency#655, count#875L])
   +- Exchange hashpartitioning(currency#655, 200), ENSURE_REQUIREMENTS, [id=#2557]
      +- HashAggregate(keys=[currency#655], functions=[partial_count(1)], output=[currency#655, count#880L])
         +- Project [currency#655]
            +- SortMergeJoin [name#0], [name#653], Inner
               :- Sort [name#0 ASC NULLS FIRST], false, 0
               :  +- Exchange hashpartitioning(name#0, 200), ENSURE_REQUIREMENTS, [id=#2549]
               :     +- Project [name#0]
               :        +- Filter isnotnull(name#0)
               :           +- Scan ExistingRDD[name#0,ticker#1,country#2,price#3L,currency#4]
               +- Sort [name#653 ASC NULLS FIRST], false, 0
                  +- Exchange hashpartitioning(name#653, 200), ENSURE_REQUIREMENTS, [id=#2550]
                     +- Project [name#653, currency#655]
                        +- Filter ((isnotnull(EPS#654) AND (cast(EPS#654 as int) > 10)) AND isnotnull(name#653))
                           +- Scan ExistingRDD[name#653,EPS#654,currency#655]

 

 

 

 

 

 

Tungsten의 기능

---

Physical Plan이 선택되면 분산 환경에서 실행되는 Java Bytecode가 만들어진다.

이 프로세스를 Code Generation이라고 한다.

 

Tungsten은 엔진의 성능 향상을 목적으로 하며, 다음과 같은 기능을 지원한다.

 

(1) 메모리 관리 최적화

(2) 캐시 활용 연산

(3) 코드 생성

 

 

 

 

 

지금까지 백엔드 컴포넌트 Catalyst, Tungsten을 알아보았다. 

이들을 공부한 이유는, 코드를 짤 때 어떻게 뒤에서 돌아갈지 상상을 해야 하기 때문이다. 이들을 이해해야 더 효율적인 프로그램을 짤 수 있을 것이다. 

다음 포스팅에선 직접 데이터를 다운로드하여 SQL 문으로 전처리를 해보자.

수고하셨습니다.