RDD 변환 (Transformations)

다음 표는 Spark이 지원하는 주요 변환(transformation)들을 정리한 거예요. 변환의 의미가 궁금하면 RDD API 문서의 Python, Scala, Java, R 그리고 페어 RDD 전용 함수 문서의 Scala, Java 페이지를 함께 보면 좋아요.

map(func)

소스 RDD의 각 원소를 함수 func에 통과시켜 만든 새로운 분산 데이터셋을 돌려줘요. 원소 하나당 결과 하나가 나오는 1:1 변환이에요.

filter(func)

func가 true를 반환하는 원소만 골라낸 새 데이터셋을 돌려줘요. 조건을 만족하지 않는 원소는 걸러지죠.

flatMap(func)

map과 비슷한데, 입력 원소 하나가 0개 이상의 출력 원소로 매핑될 수 있어요. 그래서 func는 단일 값을 반환하는 대신 Seq(시퀀스)를 반환해야 해요. 한 줄을 여러 조각으로 쪼갤 때처럼 1:N 변환에 유용하죠.

mapPartitions(func)

map과 비슷하지만, 원소 단위가 아니라 RDD의 파티션(블록) 단위로 따로 실행돼요. 그래서 RDD 타입이 T일 때 func는 반드시 Iterator<T> => Iterator<U> 타입이어야 해요. 파티션 전체를 한 번에 처리할 수 있어서 컨텍스트나 커넥션을 재사용할 때 성능상 이점이 있어요.

mapPartitionsWithIndex(func)

mapPartitions와 비슷하지만, func에 파티션의 인덱스를 나타내는 정수 값도 함께 전달해요. 그래서 RDD 타입이 T일 때 func는 (Int, Iterator<T>) => Iterator<U> 타입이어야 해요. 어느 파티션의 데이터인지 알아야 할 때 써요.

sample(withReplacement, fraction, seed)

데이터의 fraction 비율만큼 추출해요. withReplacement에 따라 복원/비복원 추출을 선택하고, 난수 생성기의 seed를 지정할 수 있어요. 재현 가능한 실험을 만들 때 seed를 고정하면 돼요.

union(otherDataset)

소스 데이터셋과 인자로 받은 데이터셋의 원소를 합집합으로 합친 새 데이터셋을 돌려줘요. 다른 RDD의 내용을 이어 붙일 때 써요.

intersection(otherDataset)

소스 데이터셋과 인자 데이터셋의 원소 중 교집합만 담은 새 RDD를 돌려줘요. 두 데이터셋에 공통으로 존재하는 원소만 남죠.

distinct([numPartitions])

소스 데이터셋의 중복을 제거한 (서로 다른 원소만 담은) 새 데이터셋을 돌려줘요.

groupByKey([numPartitions])

(K, V) 페어 데이터셋에 호출하면 (K, Iterable<V>) 페어 데이터셋을 돌려줘요. 같은 키를 가진 값들을 하나로 모아주죠.

Note: 키별로 합이나 평균 같은 **집계(aggregation)**를 하려고 그룹핑한다면, reduceByKey나 aggregateByKey를 쓰는 게 성능이 훨씬 좋아요.

Note: 기본적으로 출력의 병렬 수준은 부모 RDD의 파티션 수에 따라 결정돼요. 선택적으로 numPartitions 인자를 넘겨 작업(task) 수를 다르게 설정할 수 있어요.

reduceByKey(func, [numPartitions])

(K, V) 페어 데이터셋에 호출하면, 각 키의 값을 주어진 reduce 함수 func로 집계한 (K, V) 페어 데이터셋을 돌려줘요. func는 반드시 (V, V) => V 타입이어야 해요. groupByKey처럼 reduce 작업 수는 두 번째 인자로 조절할 수 있어요. 키별로 값을 합치고 싶을 때 제일 자주 쓰는 변환이에요.

aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])

(K, V) 페어 데이터셋에 호출하면, 각 키의 값을 주어진 결합 함수들과 중립적인 "0" 값(zero value)으로 집계한 (K, U) 페어 데이터셋을 돌려줘요. 입력 값 타입과 다른 타입의 집계 결과를 만들 수 있으면서 불필요한 할당도 피할 수 있어요. groupByKey처럼 reduce 작업 수는 두 번째 인자로 조절할 수 있어요.

sortByKey([ascending], [numPartitions])

K가 Ordered를 구현하는 (K, V) 페어 데이터셋에 호출하면, boolean ascending 인자에 따라 키를 오름차순 또는 내림차순으로 정렬한 (K, V) 페어 데이터셋을 돌려줘요.

join(otherDataset, [numPartitions])

(K, V) 타입과 (K, W) 타입 데이터셋에 호출하면, 각 키에 대해 모든 원소 쌍을 가진 (K, (V, W)) 페어 데이터셋을 돌려줘요. 외부 조인(outer join)은 leftOuterJoin, rightOuterJoin, fullOuterJoin으로 지원돼요. 두 RDD를 키 기준으로 합칠 때 쓰는 대표적인 연산이에요.

cogroup(otherDataset, [numPartitions])

(K, V) 타입과 (K, W) 타입 데이터셋에 호출하면, (K, (Iterable<V>, Iterable<W>)) 튜플 데이터셋을 돌려줘요. 이 연산은 groupWith라고도 불러요. 여러 RDD의 값을 같은 키로 묶고 싶을 때 써요.

cartesian(otherDataset)

T 타입과 U 타입 데이터셋에 호출하면, (T, U) 페어(모든 원소 쌍) 데이터셋을 돌려줘요. 두 집합의 데카르트 곱이에요.

pipe(command, [envVars])

RDD의 각 파티션을 셸 명령(예: Perl이나 bash 스크립트)에 통과시켜요. RDD 원소는 프로세스의 stdin으로 쓰여지고, stdout으로 출력된 줄들은 문자열 RDD로 반환돼요. 외부 스크립트로 데이터를 처리하고 싶을 때 써요.

coalesce(numPartitions)

RDD의 파티션 수를 numPartitions로 줄여요. 큰 데이터셋을 필터링한 뒤 작업을 더 효율적으로 실행할 때 유용해요. 셔플 없이 파티션 병합이 가능해서 repartition보다 부담이 적어요.

repartition(numPartitions)

RDD의 데이터를 무작위로 다시 섞어(reshuffle) 파티션 수를 늘리거나 줄이고 여러 파티션에 균형 있게 분배해요. 이 연산은 항상 모든 데이터를 네트워크를 통해 셔플해요. 파티션 수를 늘리거나 데이터 분포를 고르게 만들 때 써요.

repartitionAndSortWithinPartitions(partitioner)

주어진 파티셔너에 따라 RDD를 다시 분할하고, 각 결과 파티션 안에서 레코드를 키 기준으로 정렬해요. repartition을 호출한 뒤 각 파티션 안에서 정렬하는 것보다 효율적인데, 정렬을 셔플 메커니즘 안으로 밀어 넣을 수 있기 때문이에요.