Apache Spark의 parallelize는 로컬 데이터 컬렉션을 RDD(Resilient Distributed Dataset)로 변환하는 메서드이다. 이를 통해 Spark 클러스터에서 병렬 처리를 수행할 수 있다.
개요
parallelize는 Spark의 기본 RDD 생성 방법 중 하나로, 기존의 리스트 또는 배열과 같은 로컬 데이터 구조를 분산된 RDD로 변환하는 역할을 한다. 이 메서드를 활용하면 작은 데이터셋을 테스트하거나, 기존 데이터를 RDD로 변환하여 Spark에서 처리할 수 있다.
사용법
SparkContext를 사용하여 로컬 컬렉션을 RDD로 변환할 수 있다.
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder.appName("ParallelizeExample").getOrCreate()
val sc = spark.sparkContext
// 리스트 데이터를 RDD로 변환
val data = Array(1, 2, 3, 4, 5)
val rdd = sc.parallelize(data)
// RDD 데이터 출력
rdd.collect().foreach(println)
// 출력 결과
// 1
// 2
// 3
// 4
// 5
parallelize의 파티션 개수 지정
기본적으로 Spark는 클러스터의 설정에 따라 자동으로 적절한 파티션 개수를 결정하지만, 사용자가 직접 지정할 수도 있다.
val rddWithPartitions = sc.parallelize(data, numSlices = 3)
println(s"Number of Partitions: ${rddWithPartitions.getNumPartitions}")
// 출력 결과
// Number of Partitions: 3
파티션을 설정하면 Spark가 데이터를 여러 작업 노드에 나누어 병렬 처리할 수 있도록 도와준다.
parallelize 예제
기본적인 parallelize 사용
val numbers = sc.parallelize(Seq(10, 20, 30, 40, 50))
println(numbers.count()) // 결과: 5
리스트 데이터를 RDD로 변환
val words = sc.parallelize(List("Apache", "Spark", "is", "awesome"))
words.collect().foreach(println)
// 출력 결과
// Apache
// Spark
// is
// awesome
RDD 연산 적용
parallelize로 생성된 RDD에 다양한 변환 및 액션을 적용할 수 있다.
val squaredRDD = rdd.map(x => x * x)
squaredRDD.collect().foreach(println)
// 출력 결과
// 1
// 4
// 9
// 16
// 25
parallelize를 활용한 데이터 필터링
val evenRDD = rdd.filter(_ % 2 == 0)
evenRDD.collect().foreach(println)
// 출력 결과
// 2
// 4
parallelize 사용 시 주의점
- RDD는 클러스터 메모리에 존재: parallelize로 생성된 RDD는 클러스터 노드의 메모리에 저장되므로, 너무 큰 데이터를 변환하면 메모리 부족이 발생할 수 있다.
- 파티션 개수 조절 필요: 작은 데이터셋을 너무 많은 파티션으로 나누면 불필요한 오버헤드가 발생할 수 있다.
- 단일 노드에서 실행되면 성능 이점이 적음: parallelize는 Spark 클러스터에서 병렬 처리를 수행할 때 유용하지만, 단일 노드 환경에서는 성능상의 이점이 크지 않을 수 있다.