[Spark] Spark 속도 최적화, Cashe() & Persist()

 

 

Spark 속도를 최적화시키는 방법으론,

Cashe(), Persist()가 있다.

 

 

 Spark 특성상, Transformations는 Lazy Execution 방식으로 작업을 수행한다.

그 이유는, Eager Execution(즉시 시행)을 따를 경우 다음과 같은 비효율이 발생한다.

 

 

 

 

 

 

 

Task를 수행할 때마다 Disk에 저장을 하고, 다시 이를 불러들여야 한다.

이 과정에서 Disk 공간은 물론, 통신으로 인한 속도 저하 등 비효율이 발생하게 된다.

그렇다면 Lazy Execution 방식으로 작업을 수행한다면 어떨까?

 

 

 

 

 

 

 

Lazy Execution 방식이라면, Disk에 저장하지 않는다. 대신 Task에서 Task로 전달될 때 In-Memory 방식으로 데이터를 주고받을 수 있다. In-Memory 방식으로 주고받기 위해 사용하는 것이 바로, Cashe(), Persist() 이다.

코드를 통해 바로 실습해보자.

 

 

 

 

 

conf 설정

 

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

conf = SparkConf().setMaster('local').setAppName('my-RDD-transformation-action')
sc = SparkContext(conf=conf)

# RDD 생성
foods = sc.parallelize(['짜장면', '짬뽕', '볶음밥', '떡볶이', '라면', '우동', '짜장면', '짬뽕', '볶음밥', '떡볶이', '라면', '우동', '짜장면', '짬뽕', '볶음밥'])

 

 

 

 

 

Code

---

 

Cache(), Persist()를 활용하지 않은 경우.

 

def parse(row):
    return (row, 1)

key_foods = foods.map(parse)
key_foods.collect()


[('짜장면', 1),
 ('짬뽕', 1),
 ('볶음밥', 1),
 ('떡볶이', 1),
 ('라면', 1),
 ('우동', 1),
 ('짜장면', 1),
 ('짬뽕', 1),
 ('볶음밥', 1),
 ('떡볶이', 1),
 ('라면', 1),
 ('우동', 1),
 ('짜장면', 1),
 ('짬뽕', 1),
 ('볶음밥', 1)]

 

map() 함수를 활용해서 key_foods RDD를 만들었다. 

 

result1 = key_foods.take(3)
result2 = key_foods.mapValues(lambda x: x + 1).collect()
print(result2)


[('짜장면', 2), ('짬뽕', 2), ('볶음밥', 2), 
('떡볶이', 2), ('라면', 2), ('우동', 2), ('짜장면', 2), 
('짬뽕', 2), ('볶음밥', 2), ('떡볶이', 2), ('라면', 2), 
('우동', 2), ('짜장면', 2), ('짬뽕', 2), ('볶음밥', 2)]

 

Cache(), Persist() 함수를 사용하지 않았기에, 위의 코드에선 key_foods가 두 번 만들어진다. take() 함수를 실행하면서 1회 만들어지고, mapValues를 실행하면서 또다시 한번 만들어지는 것이다. 이러한 비효율은 속도를 저하시킬 수 있다. 

 

 

 

 

Cache(), Persist()를 활용한 경우.

 

def parse(row):
    return (row, 1)

key_foods = foods.map(parse).persist()
key_foods.collect()


[('짜장면', 1),
 ('짬뽕', 1),
 ('볶음밥', 1),
 ('떡볶이', 1),
 ('라면', 1),
 ('우동', 1),
 ('짜장면', 1),
 ('짬뽕', 1),
 ('볶음밥', 1),
 ('떡볶이', 1),
 ('라면', 1),
 ('우동', 1),
 ('짜장면', 1),
 ('짬뽕', 1),
 ('볶음밥', 1)]

 

map() 함수를 적용하고, 곧바로 persist() 함수를 이용해 메모리에 저장하도록 했다.

 

result1 = key_foods.take(3)
result2 = key_foods.mapValues(lambda x: x + 1).collect()
print(result2)


[('짜장면', 2), ('짬뽕', 2), ('볶음밥', 2), ('떡볶이', 2), 
('라면', 2), ('우동', 2), ('짜장면', 2), ('짬뽕', 2), 
('볶음밥', 2), ('떡볶이', 2), ('라면', 2), ('우동', 2), 
('짜장면', 2), ('짬뽕', 2), ('볶음밥', 2)]

 

이제는 key_foods 연산이 한 번만 실행되고, 메모리에 있던 key_foods가 활용되기만 할 뿐이다. 

 

 

 

 

이외에도 머신러닝 학습같이, 데이터를 다루는 task가 반복되는 경우가 많을 때도 활용할 수 있다. 

 

 

출처: 위키백과

 

points = sc.textFile('...').map(parsePoint).cache()

for i in range(ITERATIONS):
	gradient = points.map(gradient_descent).reduce(lambda x, y: (x + y) / n)
    w -= gradient * learning_rate

cache() 함수를 이용하지 않았다면, gradient를 계산할 때마다 points를 연산하는 끔찍한 상황이 연출되었을 것이다. 

 

 

 

다음은 Cache(), Persist()를 활용할 때, 참고사항이다. 

 

Storage Level

다양한 스토리지 레벨이 존재한다. 사용자는 특정 레벨을 지정해서 활용할 수 있다.

1) MEMORY_ONLY - 메모리에만 저장

2) MEMORY_AND_DISK - 메모리와 디스크에 저장

3) MEMORY_ONLY_SER - 메모리에만 시리얼 라이져 시킨 다음 저장

4) MEMORY_AND_DISK_SER - 메모리와 디스크에 시리얼라이져 시킨 다음 저장

5) DISK_ONLY - 디스크에만 저장

 

 

Cache

Cache()의 설정 옵션이다.

1) Default Storage Level 사용

2) RDD - MEMORY_ONLY

3) DataFrame - MEMORY_AND_DISK

 

 

Persist

Persist()의 설정 옵션이다.

1) Storage Level을 사용자가 원하는 대로 지정 가능