RDD Transformations

상위: Spark

요약

RDD Transformation은 기존 RDD를 새로운 RDD로 변환하는 연산입니다. map, filter, flatMap 등의 기본 연산부터 groupBy, join 등의 집계 연산까지 다양하며, Lazy Evaluation 방식으로 동작합니다.

MAP

각 요소에 함수를 적용하여 1:1 변환

x = sc.parallelize(['b', 'a', 'c'])
y = x.map(lambda z: (z, 1))
print(x.collect())
print(y.collect())

출력:

x: ['b', 'a', 'c']
y: [('b', 1), ('a', 1), ('c', 1)]

FLATMAP

각 요소를 여러 요소로 확장 (1:N 변환)

x = sc.parallelize([1, 2, 3])
y = x.flatMap(lambda x: (1*x, 2*x, 3*x, 100))
print(x.collect())
print(y.collect())
print(y.mean())

출력:

x: [1, 2, 3]
y: [1, 2, 3, 100, 2, 4, 6, 100, 3, 6, 9, 100]
y.mean(): 28.0

FILTER

조건을 만족하는 요소만 필터링

x = sc.parallelize([1, 2, 3])
y = x.filter(lambda x: x % 2 == 1)  # 홀수만 필터링
print(x.collect())
print(y.collect())

출력:

x: [1, 2, 3]
y: [1, 3]

MAPPARTITIONS

파티션 단위로 함수를 적용

x = sc.parallelize([1, 2, 3], 2)

def f(iterator):
    yield sum(iterator)
    yield 42

y = x.mapPartitions(f)

print(x.glom().collect())  # 각 파티션의 원소 확인
print(y.glom().collect())  # 파티션별 결과

출력:

x: [[1], [2, 3]]
y: [[1, 42], [5, 42]]

MAPPARTITIONS WITH INDEX

파티션 인덱스와 함께 처리

x = sc.parallelize([1, 2, 3], 2)

def f(partitionIndex, iterator):
    yield (partitionIndex, sum(iterator))

y = x.mapPartitionsWithIndex(f)

print(x.glom().collect())
print(y.glom().collect())

출력:

x: [[1], [2, 3]]
y: [[(0, 1)], [(1, 5)]]

KEYBY

각 요소에서 키를 추출하여 (key, value) 형태로 변환

x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])

y = x.keyBy(lambda w: w[0])

print(y.collect())

출력:

x: ['John', 'Fred', 'Anna', 'James']
y: [('J', 'John'), ('F', 'Fred'), ('A', 'Anna'), ('J', 'James')]

GROUPBY

키를 기준으로 그룹화

x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])

y = x.groupBy(lambda w: w[0])
print([{t[0]: [i for i in t[1]]} for t in y.collect()])

출력:

x: ['John', 'Fred', 'Anna', 'James']
y: [{'A': ['Anna']}, {'J': ['John', 'James']}, {'F': ['Fred']}]

GROUPBYKEY

(key, value) RDD를 키로 그룹화

x = sc.parallelize([('B', 5), ('B', 4), ('A', 3), ('A', 2), ('A', 1)])

y = x.groupByKey()
print(x.collect())
print([(t[0], [i for i in t[1]]) for t in y.collect()])

출력:

x: [('B', 5), ('B', 4), ('A', 3), ('A', 2), ('A', 1)]
y: [('B', [5, 4]), ('A', [3, 2, 1])]

Word Count using GROUPBYKEY

words = sc.parallelize(['one', 'two', 'two', 'three', 'three', 'three'])
wordPairsRdd = words.map(lambda w: (w, 1))
wordCounts = wordPairsRdd.groupByKey().map(lambda pair: (pair[0], sum(pair[1])))

print(words.collect())
print(wordPairsRdd.collect())
print(wordCounts.collect())

출력:

words: ['one', 'two', 'two', 'three', 'three', 'three']
wordPairsRDD: [('one', 1), ('two', 1), ('two', 1), ('three', 1), ('three', 1), ('three', 1)]
wordCounts: [('one', 1), ('two', 2), ('three', 3)]

REDUCEBYKEY

Grouping + Aggregation을 한 번에 수행

words = sc.parallelize(['one', 'two', 'two', 'three', 'three', 'three'])
wordPairsRdd = words.map(lambda w: (w, 1))
wordCounts = wordPairsRdd.reduceByKey(lambda cnt1, cnt2: cnt1 + cnt2)

print(words.collect())
print(wordPairsRdd.collect())
print(wordCounts.collect())

출력:

words: ['one', 'two', 'two', 'three', 'three', 'three']
wordPairsRDD: [('one', 1), ('two', 1), ('two', 1), ('three', 1), ('three', 1), ('three', 1)]
wordCounts: [('one', 1), ('two', 2), ('three', 3)]

REDUCEBYKEY vs GROUPBYKEY

• 두 함수가 모두 사용 가능하다면, ReduceByKey 사용 권장
• ReduceByKey는 셔플 전에 행을 결합하여 셔플해야 할 행의 수를 줄일 수 있음
  • 로컬 집계 수행
  • 중간 결과의 크기를 줄일 수 있음

REDUCEBYKEY: 집계하고 Shuffle → 효율적

GROUPBYKEY: Shuffle하고 집계 → 비효율적

JOIN

두 RDD를 키 기준으로 조인

x = sc.parallelize([("a", 1), ("b", 2)])
y = sc.parallelize([("a", 3), ("a", 4), ("b", 5)])

z = x.join(y)
print(z.collect())

출력:

x: [("a", 1), ("b", 2)]
y: [("a", 3), ("a", 4), ("b", 5)]
z: [("a", (1, 3)), ("a", (1, 4)), ("b", (2, 5))]

UNION

두 RDD를 합침 (중복 제거 안 함)

x = sc.parallelize([1, 2, 3], 2)
y = sc.parallelize([3, 4], 1)
z = x.union(y)

print(z.glom().collect())

출력:

x: [[1], [2, 3]]
y: [3, 4]
z: [[1], [2, 3], [3, 4]]

DISTINCT

중복을 제거 (Shuffle 발생)

x = sc.parallelize([1, 2, 3, 3, 4])
y = x.distinct()

print(y.collect())

출력:

x: [1, 2, 3, 3, 4]
y: [1, 2, 3, 4]

SAMPLE

데이터를 샘플링

형식: sample(withReplacement, fraction, seed=None)

x = sc.parallelize([1, 2, 3, 4, 5])
y = x.sample(False, 0.4, 42)

print(x.collect())
print(y.collect())

출력:

x: [1, 2, 3, 4, 5]
y: [1, 3]

COALESCE

파티션 수를 줄임 (Shuffle 없이)

x = sc.parallelize([1, 2, 3, 4, 5], 3)
y = x.coalesce(2)

print(x.glom().collect())
print(y.glom().collect())

출력:

x: [[1], [2, 3], [4, 5]]
y: [[1], [2, 3, 4, 5]]

COALESCE vs REPARTITION

• COALESCE: 파티션 줄이기 (merge), Shuffle 없음
• REPARTITION: 파티션 늘리기 or 균등 재분배, Shuffle 발생

PARTITIONBY

커스텀 파티셔너로 재분배

x = sc.parallelize([('J', 'James'), ('F', 'Fred'), ('A', 'Anna'), ('J', 'James')], 3)
y = x.partitionBy(2, lambda w: 0 if w[0] < 'H' else 1)

print(x.glom().collect())
print(y.glom().collect())

출력:

x: [[('J', 'James')], [('F', 'Fred'), ('A', 'Anna')], [('J', 'James')]]
y: [[('A', 'Anna'), ('F', 'Fred')], [('J', 'James'), ('J', 'James')]]