repartition vs coalesce



Repartition
Coalesce
Increasing and decreasing the number of partitions
Can increase and decrease the number of partitions
Can only decrease the number of partitions
Data movement
More, as it does full shuffle and creates new partitions
Less; Avoids full shuffle, by combining the existing partitions.
Example
RDD is representing the data on 4 nodes:

Node 1: a,b,c,d
Node 2: e,f,g,h
Node 3: i,j,k,l
Node 4: m,n

Each node’s data represents a partition.

After running the repartition to make two partitions of the resultant data, it could become:

Node 5: a,b,f,h,i,j,n
Node 6: c,d,e,g,k,l,m

The new distribution is random(or evenly moved)
RDD is representing the data on 4 nodes:

Node 1: a,b,c,d
Node 2: e,f,g,h
Node 3: i,j,k,l
Node 4: m,n

Each node’s data represents a partition.

After running coalesce to make into two partitions, it could become
Node 1: a,b,c,d,i,j,k,l
Node 2: e,f,g,h,m,n


Node 1 and node 2’s data didn’t go through the data movement
Resultant
New partition
Uses existing partition
Partition sizes after repartitions
Roughly equal sized partitions,
Different amounts of data
Speed
Overall repartition is faster because spark is built into work with equal sized partitions.
May run faster than repartition, but unequal sized are generally slower than equal sized partitions
Original partition
Will not be modified
Will not be modified
Internal calls
Repartition calls coalesce with shuffling set to true
This is a direct call
Data distribution
Equal distribution in the resultant partitions
Doesn’t follow the equal distribution in the resultant partitions.
What if we try to increase with coalesce?

No error will be thrown, it just ignores.

If we call coalesce on the repartitioned data it works.- why??
Table partition
val colorsRDD = List(
  (10, "green"),
  (13, "red"),
  (15, "blue"),
  (99, "red"),
  (67, "Yellow")
)

//Converting to DF
val convertedDF = colorsRDD.toDF("ID", "color")
//Repartitioning
resultantRepartition = convertedDF.repartition($"color")

ð  Creates partitions for each color.

When works well?

RDD with lot of partitions and combining the partitions on a single worker node to produce the final resultant partition.
         

Note: Repartition is always a costly operation.
Table partitions, spark by default creates 200 partitions.
Partitioning table by a column is similar to indexing a column in RDBMS

Comments

Post a Comment

Popular posts from this blog

Out Of Memory in Spark(OOM) - Typical causes and resolutions

   Spark OOM – Can occur at Driver/Executor/node manager. The error that we see is: java.lang.OutOfMermoryError. This will happen when the required memory exceeds the available memory for a particular operation. 1.        Driver OOM - possible causes: a.        Collect For e.g. in the program, we have    val data = inDF.collect()                 Collect operation, collects the data from all executors and send it to the driver. The driver will try to merge into the single object, then there is a possibility that it may be too big to fit into the driver memory. This problem can be solved in two ways: Ø   Setting proper limit using spark.driver.maxResultSize Ø   Repartitioning before saving the result to the output file. §   data.repartition(1).write… §   Repartition uses dedicated executor to collect. b.        Broadcast Join ¨        The table will be materialized at the driver and then broadcasted to the exectuors. c.        Low driver memory configurati
Read more

map vs flatMap in Spark

Image
map flatMap Returns single element based on the function/custom business logic/algorithm Returns zero or more elements based on the function/custom business logic/algorithm Single element as return type Iterator with our return values, but we don’t return an iterator of RDD’s, return RDD that consists of the elements from all the iterators. e.g. val list = 1 to 5 toList list: List[Int] = List(1, 2, 3, 4, 5) list.map(_.to(3)) List[scala.collection.immutable.Range.Inclusive] = List(Range(1, 2, 3), Range(2, 3), Range(3), Range(), Range()) E.g. val list = 1 to 5 toList list: List[Int] = List(1, 2, 3, 4, 5) list.flatMap(_.to(3)) List[Int] = List(1, 2, 3, 2, 3, 3)   Returns  RDD of lists ; refer to the example Returns  RDD of elements map operation only Equivalent to map followed by flatten Returns the elements of the RDD Returns the elements of the I
Read more

Spark Persistence(Caching)

Image
Cached RDDs should have modest size so that they can fit in the MEMORY entirely. To identify the size, it is challenging and unclear. Caching strategy that caches the blocks in memory and disk is preferred. The reason is the cached blocks that are evicted will be written to disk. Reading from disk is relatively fast as compared to re-evaluating the RDD. Internals: Internally caching is performed at the block level. That means each RDD consists of multiple blocks, each block is being cached independently of other blocks. Caching is performed on the node that is generated the particular RDD block. Each executor in spark has an associated Block Manager that is used to cache RDD blocks. The memory allocation for the Block Manager is given by the storage memory fraction, which gives the fraction from the memory pool that is allocated for the spark engine. The block manager manages
Read more