9

Does flatMap in spark behave like the map function and therefore cause no shuffling, or does it trigger a shuffle. I suspect it does cause shuffling. Can someone confirm it?

Community
  • 1
  • 1
pythonic
  • 20,589
  • 43
  • 136
  • 219
  • 2
    See http://stackoverflow.com/questions/26273664/what-are-the-spark-transformations-that-causes-a-shuffle – Alexis C. Apr 04 '16 at 23:10

3 Answers3

8

There is no shuffling with either map or flatMap. The operations that cause shuffle are:

  • Repartition operations:
    • Repartition:
    • Coalesce:
  • ByKey operations (except for counting):
    • GroupByKey:
    • ReduceByKey:
  • Join operations:
    • Cogroup:
    • Join:

Although the set of elements in each partition of newly shuffled data will be deterministic, and so is the ordering of partitions themselves, the ordering of these elements is not. If one desires predictably ordered data following shuffle then it’s possible to use:

  • mapPartitions to sort each partition using, for example, .sorted
  • repartitionAndSortWithinPartitions to efficiently sort partitions while simultaneously repartitioning
  • sortBy to make a globally ordered RDD

More info here: http://spark.apache.org/docs/latest/programming-guide.html#shuffle-operations

JorgeGlezLopez
  • 106
  • 2
  • The answer needs a correction- Coalesce doesnt produce a shuffle. It is a narrow rdd dependency. – human Nov 09 '17 at 00:39
6

No shuffling. Here are the sources for both functions:

/**
 * Return a new RDD by applying a function to all elements of this RDD.
 */
def map[U: ClassTag](f: T => U): RDD[U] = withScope {
  val cleanF = sc.clean(f)
  new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
}

/**
 *  Return a new RDD by first applying a function to all elements of this
 *  RDD, and then flattening the results.
 */
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] = withScope {
  val cleanF = sc.clean(f)
  new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.flatMap(cleanF))
}

As you can see, RDD.flatMap just calls flatMap on Scala's iterator that represents partition.

Aivean
  • 10,692
  • 25
  • 39
0

flatMap may cause shuffle write in some cases. like if you are generating multiple elements into the same partition and that element can't fit into the same partition then it writes those into a different partition.

like in below example : 

val rdd = RDD[BigObject]

rdd.flatMap{ bigObject => 
    val rangList: List[Int] = List.range(1, 1000)
    rangList.map( num => (num, bigObject))
}

Above code will run on the same partition but since we are creating too many instances of BigObject , it will write those objects into separate partitions which will cause shuffle write

Darshan Butiya
  • 1
  • 1