Spark : Internals of Spark Job

It is very important to understand the internals of a Spark job, like what are the stages involved when we run a Spark Job etc. It will help understand the performance of the Job and we can decide on which particular steps are impacting performance and we can exclude such steps incase if they are not needed.

Our entire job will be divided into stages, and there are 2 types of dependencies. We need to understand how stages will come, what is the importance of those stages and what are the dependencies. All these are interconnected.

When we trigger some action, a Spark Job will be started.

Spark Job 

• Stages
• Tasks (every stage will have set of tasks which execute in parallel)
• Dependencies
• Narrow dependency
• Shuffle dependency

    

Lets understand what are the dependencies. Ideally, below are 4 types of mappings for any actions in Spark.

1. One to One (one element to another element - each element is independent)
2. One to Many (This is also independent on each element)
3. Many to One (This is dependent on one another)
4. Many to Many (This is dependent on one another)

Important point is : 

• One to One, One to Many comes under Narrow dependency.
• Many to One, Many to Many comes under Shuffle dependency

For your understanding, filter(), map(),  union() operation is 1-1/1-many which comes under Narrow dependency BUT sortBy(), groupBy() comes under Many -1/Many-Many which comes under Shuffle dependency.

Let's say, we have below tasks going to execute one after another(N is Narrow, S is Shuffle) :

N,    N,    N,                    S,    N,    N,                    S,                    S

Whenever there is a task which involves a task which falls under Shuffle dependency, then a new Stage will start. From above list of tasks, there will be 4 different stages created on that Spark Job.

Please see below Spark Job :

// Created a RDD, r1

scala> val r1 = sc.parallelize(1 to 6, 2)

r1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[23] at parallelize at <console>:23

// From above RDD r1, we created 2 other RDD's r11, r12

scala> val r11 = r1.filter(x => x % 2 == 0)

r11: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[24] at filter at <console>:23

scala> val r12 = r1.filter(x => x % 2 == 1)

r12: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[25] at filter at <console>:23

// Again, created 2 more RDD's from each of r11, r12

scala> val r111 = r11.map(x => x + 1)

r111: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[26] at map at <console>:23

scala> val r121 = r12.map(x => x + 1)

r121: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[27] at map at <console>:23

// Now using union() and combining 2 RDD's r111, r121

scala> val r2 = r111.union(r121)

r2: org.apache.spark.rdd.RDD[Int] = UnionRDD[28] at union at <console>:24

// Now we are using sort() which comes under Shuffle Dependency, hence new stage will start here

// Understand that in sort() operation elements are interdependent on each other, hence costly

scala> val r3 = r2.sortBy(x => x)

r3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[33] at sortBy at <console>:23

// Using map() which is a Narrow dependency

scala> val r31 = r3.map(x => x + 1)

r31: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[34] at map at <console>:23

// Using filter() which is also a Narrow dependency

scala> val r311 = r31.filter(x => x % 2 == 1)

r311: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[35] at filter at <console>:23

// Here, groupBy() comes under Shuffle dependency as elements are interdependent

// Hence a new stage will be created

scala> val r4 = r311.groupBy(x => x % 2 == 0)

r4: org.apache.spark.rdd.RDD[(Boolean, Iterable[Int])] = ShuffledRDD[37] at groupBy at <console>:23

// Finally printing RDD r4

scala> r4.collect()

res4: Array[(Boolean, Iterable[Int])] = Array((false,CompactBuffer(3, 5, 7)))

Now, from above Spark Job, Spark should create 3 different stages. We can visualize it from our local system using URL : http://localhost:4040/jobs/ 

From same URL, we can see the Tasks as well as shown in below image.

We need to explore and understand more about Job, Stage, Dependencies, Tasks to under architecture. We can use local system incase if you have environment setup to execute Spark jobs, or else please use Databricks community addition by creating an account and explore.

toDebugString : Observed when we use toDebugString, it is displaying the information of all 3 stages.

scala> r4.toDebugString

res5: String =

(4) ShuffledRDD[37] at groupBy at <console>:23 []

 +-(4) MapPartitionsRDD[36] at groupBy at <console>:23 []

    |  MapPartitionsRDD[35] at filter at <console>:23 []

    |  MapPartitionsRDD[34] at map at <console>:23 []

    |  MapPartitionsRDD[33] at sortBy at <console>:23 []

    |  ShuffledRDD[32] at sortBy at <console>:23 []

    +-(4) MapPartitionsRDD[29] at sortBy at <console>:23 []

       |  UnionRDD[28] at union at <console>:24 []

       |  MapPartitionsRDD[26] at map at <console>:23 []

       |  MapPartitionsRDD[24] at filter at <console>:23 []

       |  ParallelCollectionRDD[23] at parallelize at <console>:23 []

       |  MapPartitionsRDD[27] at map at <console>:23 []

       |  MapPartitionsRDD[25] at filter at <console>:23 []

       |  Par...

scala> r4.dependencies

res6: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.ShuffleDependency@4bc0f78e)

dependencies will give what is the dependency on that RDD.

That's all for this blog. See you again!

Thanks,

Arun Mathe

Email ID : arunkumar.mathe@gmail.com