Partitioning in Apache Spark

Data in the same partition will always be in the same machine. Data in a partition will not span multiple machines.

Default no of partition is equal to the number of CPU core in the machine.

partitioning can bring a substantial performance gain by reducing the amount of data to be shuffled across the network.

Spark can run 1 concurrent task for every partition of an RDD . In general, more numerous partitions allow work to be distributed among more workers and achieve better parallelism but fewer partitions allow work to be done in larger chunks (and often quicker if task scheduling may take more time than actual execution time).

How to Choose Number of Partitions :

• Lower bound — 2 X number of cores in cluster available to application
• Upper bound — task should take 100+ ms time to execute.If it is taking less time than your partitioned data is too small and your application might be spending more time in scheduling the tasks.

Two types of partitioning:

1. Hash Partitioning

It spreads around the data in the partitioning based upon the key value.

p=key.hashCode() %noOfPartitions

Hash partitioning can make distributed data skewed.

2. Range Partitioning.

It partition data either based on some sorted order OR set of sorted ranges of keys, tuples with the same range will be on the same machine.

Partitioning is only possible in pair RDDs.

Performance gain when data was partitioned before applying the transformation.

https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/best_practices/prefer_reducebykey_over_groupbykey.html

How to partition your RDD

1. By calling the partitionBy method and passing partition function.

val pairRDD = data.map(x =>(x.key,x.value));
val partitioner  = new RangePartitioner(8,pairRDD) ;
val partitionedRDD = pairRDD.partitionBy(partitioner).persist();

To create a range partitioner we require number of partition count and the pairRDD which will be sampled to create suitable data range for partitioned data.

persist should be called if you want to perform more operation on suffled/partitioned RDD.

2. By calling transformation that returns RDD using specific partitioners.

Spark knows internally how each of its operations affects partitioning, and automatically sets the partitioner on RDDs created by operations that partition that data.

GroupByKey ,ReduceByKey — by default this operation uses Hash Partitioning with default parameters.

SortByKey-Range partitioning

But there are some transformations that cannot guarantee to produce known partitioning — for example calling map() could theoretically modify the key of each element.

Instead, there are some functions provided that guarantee that each tuple’s key remains the same — mapValues(), flatMapValues() or filter() (if the parent has a partitioner).

In case of parallelize operation data is evenly distributed between partitions using indices. So default partitioning scheme is simply none because partitioning is not applicable to all RDDs.

Why partition data?

partition helps in localizing the data and reduce the data shuffling across the network nodes reducing network latency which is a major component of the transformation operation thereby reducing the time of completion.

A scenario where we want to perform more than one operation based upon keys like countByKey, groupByKey etc.

Figure 3

Figure 3 suboptimal case multiple transformations are done on the same original RDD which require shuffling of the data multiple time across the network.

To improve on this performance we can create a partitioned RDD and perform shuffle transformation operation over that persisted RDD which reduces multiple times shuffling of the data