[Spark] Reduction 개념 및 코드

 

 

Reduction

요소들을 모아서 하나로 합치는 작업을 말한다.

많은 Spark 연산들이 Reduction이라고 봐도 무방하다.

* 파일 저장, collect() 등과 같이 Reduction이 아닌 액션도 존재.

코드로 실습해보자.

 

 

 

 

 

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)

 

 

 

 

 

대표적인 Reduction Actions

---

 

1) Reduction

 

RDD.reduce(function)

 

from operator import add
sc.parallelize([1,2,3,4,5]).reduce(add)


15

 

 

add 함수를 이용하여 데이터를 합쳤다.

Partition에 따라 결과 값이 달라지기도 한다. 아래 코드를 보자. 

 

 

 

# Partition - 1
sc.parallelize([1,2,3,4], 1).reduce(lambda x,y: (x*2) + y)


26

----
# Partition - 2
sc.parallelize([1,2,3,4], 2).reduce(lambda x,y: (x*2) + y)


18

----
# Partition - 3
sc.parallelize([1,2,3,4], 3).reduce(lambda x,y: (x*2) + y)


18

 

 

이처럼 파티션이 어떻게 나뉘냐에 따라 결과 값이 다를 수 있으므로, 신경 써줘야 한다.

 

 

 

 

2) Fold

 

RDD.fold(zeroValue, function)

 

 

from operator import add
sc.parallelize([1,2,3,4,5]).fold(0, add)


15

 

 

reduce와 마찬가지로 데이터를 합치는 함수이지만, 양상이 살짝 다르다.

기본 값과 RDD를 같이 function해주는 함수다. 

 

 

rdd = sc.parallelize([2,3,4], 4)
rdd.fold(1, lambda x, y: x+y)


# (1+1) + (1+2) + (1+3) + (1+4) = 14
14

 

 

다른 활용 예시

 

 

rdd = sc.parallelize([2,3,4], 4)
rdd.fold(3, lambda x, y: x*y)


# (3*3) * (3*2) * (3*3) * (3*4) = 5832
5832

 

 

 

 

3) GroupBy

 

RDD.groupBy(function)

 

 

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
rdd_group = rdd.groupBy(lambda x: x % 2).collect()
print(rdd_group)


[(1, <pyspark.resultiterable.ResultIterable at 0x134519253c8>),
 (0, <pyspark.resultiterable.ResultIterable at 0x13452665ec8>)]
 
 ----
 
 list(rdd_group[0][1])
 
 
 [1, 3, 5, 7, 9]

 

 

연산 기준을 통해 데이터를 grouping하는 시행이다.

 

 

 

 

4) Aggregate

 

RDD.aggregate(zeroValue, seqOp, combOp)

- zeroValue: 각 파티션에서 누적할 시작 값

- seqOp: 타입 변경 함수

- combOp: 합치는 함수

 

 

seqOp = (lambda x,y: (x[0] + y, x[1] + 1))
combOp = (lambda x,y: (x[0] + y[0], x[1] + y[1]))
sc.parallelize([1,2,3,4,5]).aggregate((0,0), seqOp, combOp)


(15,5)

 

 

많이 쓰이는 reduction action.

대부분의 데이터 작업은 크고 복잡한 데이터 타입 -> 정제된 데이터로 변환.