Apache Spark - Best Practices and Tuning
Search…
Introduction
RDD
Don’t collect large RDDs
Don't use count() when you don't need to return the exact number of rows
Avoiding Shuffle "Less stage, run faster"
Picking the Right Operators
TreeReduce and TreeAggregate Demystified
When to use Broadcast variable
Joining a large and a small RDD
Joining a large and a medium size RDD
Dataframe
Joining a large and a small Dataset
Joining a large and a medium size Dataset
Storage
Use the Best Data Format
Cache Judiciously and use Checkpointing
Parallelism
Use the right level of parallelism
How to estimate the size of a Dataset
How to estimate the number of partitions, executor's and driver's params (YARN Cluster Mode)
Serialization and GC
Tuning Java Garbage Collection
Serialization
References
References
Powered By GitBook
Joining a large and a medium size RDD
If the medium size RDD does not fit fully into memory but its key set does, it is possible to exploit this [23]. As a join will discard all elements of the larger RDD that do not have a matching partner in the medium size RDD, we can use the medium key set to do this before the shuffle. If there is a significant amount of entries that gets discarded this way, the resulting shuffle will need to transfer a lot less data.
val keys = sc.broadcast(mediumRDD.map(_._1).collect.toSet)
val reducedRDD = largeRDD.filter{ case(key, value) => keys.value.contains(key) }
reducedRDD.join(mediumRDD)
It is important to note that the efficiency gain here depends on the filter operation actually reducing the size of the larger RDD. If there are not a lot of entries lost here (e.g., because the medium size RDD is some king of large dimension table), there is nothing to be gained with this strategy.
You can find more details for efficient shuffles in this Databricks presentation.
RDD - Previous
Joining a large and a small RDD
Next - Dataframe
Joining a large and a small Dataset
Last modified 2yr ago
Copy link