Don’t collect large RDDs

Apache Spark - Best Practices and Tuning

Search…

When a collect operation is issued on a RDD, the dataset is copied to the driver, i.e. the master node. A memory exception will be thrown if the dataset is too large to fit in memory; takeor takeSamplecan be used to retrieve only a capped number of elements instead.
Another way has been showed in [8] where you can get the array of partition indexes:
1
val parallel = sc.parallelize(1 to 9)
2
val parts = parallel.partitions
Copied!
and then create a smaller rdd filtering out everything but a single partition. Collect the data from smaller rdd and iterate over values of a single partition:
1
for(p <- parts){
2
  val idx = p.index
3
  val partRDD = parallel.mapPartitionsWithIndex((index: Int, it: Iterator[Int]) => if(index == idx) it else Iterator(), true)
4
  val data = partRDD.collect
5
  // Data contains all values from a single partition in the form of array.
6
  // Now you can do with the data whatever you want: iterate, save to a file, etc.
7
}
8
​
9
// You can use also the foreachPartition operation
10
parallel.foreachPartition(partition => {
11
  partition.toArray
12
  // Your code
13
})
Copied!
Of cause, it will work only if the partitions are small enough. If they aren't, you can always increase the number of partitions with rdd.coalesce(numParts, true).

Introduction

Next - RDD

Don't use count() when you don't need to return the exact number of rows

Last modified 2yr ago

Copy link