Spark SQL: Why two jobs for one query?
<p><strong>Experiment</strong></p>

<p>I tried the following snippet on <code>Spark 1.6.1</code>.</p>

<pre><code>val soDF = sqlContext.read.parquet("/batchPoC/saleOrder") # This has 45 files
soDF.registerTempTable("so")
sqlContext.sql("select dpHour, count(*) as cnt from so group by dpHour order by cnt").write.parquet("/out/")
</code></pre>

<p>The <code>Physical Plan</code> is:</p>

<pre><code>== Physical Plan ==
Sort [cnt#59L ASC], true, 0
+- ConvertToUnsafe
   +- Exchange rangepartitioning(cnt#59L ASC,200), None
      +- ConvertToSafe
         +- TungstenAggregate(key=[dpHour#38], functions=[(count(1),mode=Final,isDistinct=false)], output=[dpHour#38,cnt#59L])
            +- TungstenExchange hashpartitioning(dpHour#38,200), None
               +- TungstenAggregate(key=[dpHour#38], functions=[(count(1),mode=Partial,isDistinct=false)], output=[dpHour#38,count#63L])
                  +- Scan ParquetRelation[dpHour#38] InputPaths: hdfs://hdfsNode:8020/batchPoC/saleOrder
</code></pre>

<p>For this query, I got two Jobs: <code>Job 9</code> and <code>Job 10</code>
<a href="https://i.stack.imgur.com/pPD3P.jpg" rel="noreferrer"><img src="https://i.stack.imgur.com/pPD3P.jpg" alt="enter image description here"></a></p>

<p>For <code>Job 9</code>, the <code>DAG</code> is:</p>

<p><a href="https://i.stack.imgur.com/Gnnvn.jpg" rel="noreferrer"><img src="https://i.stack.imgur.com/Gnnvn.jpg" alt="enter image description here"></a></p>

<p>For <code>Job 10</code>, the <code>DAG</code> is:</p>

<p><a href="https://i.stack.imgur.com/qpvhI.jpg" rel="noreferrer"><img src="https://i.stack.imgur.com/qpvhI.jpg" alt="enter image description here"></a></p>

<p><strong>Observations</strong></p>

<ol>
<li>Apparently, there are two <code>jobs</code> for one query. </li>
<li><code>Stage-16</code> (marked as <code>Stage-14</code> in <code>Job 9</code>) is skipped in <code>Job 10</code>.</li>
<li><code>Stage-15</code>'s last <code>RDD[48]</code>, is same as <code>Stage-17</code>'s last <code>RDD[49]</code>. <em>How</em>? I saw in the logs that after <code>Stage-15</code> execution, the <code>RDD[48]</code> is registered as <code>RDD[49]</code></li>
<li><code>Stage-17</code> is shown in the <code>driver-logs</code> but never got executed at <code>Executors</code>. On <code>driver-logs</code> the task-execution is shown, but when I looked at <code>Yarn</code> container's logs, there was no evidence of receiving any <code>task</code> from <code>Stage-17</code>.  </li>
</ol>

<p>Logs supporting these observations (only <code>driver-logs</code>, I lost <code>executor</code> logs due to later crash). It is seen that before <code>Stage-17</code> starts, <code>RDD[49]</code> is registered:</p>

<pre><code>16/06/10 22:11:22 INFO TaskSetManager: Finished task 196.0 in stage 15.0 (TID 1121) in 21 ms on slave-1 (199/200)
16/06/10 22:11:22 INFO TaskSetManager: Finished task 198.0 in stage 15.0 (TID 1123) in 20 ms on slave-1 (200/200)
16/06/10 22:11:22 INFO YarnScheduler: Removed TaskSet 15.0, whose tasks have all completed, from pool 
16/06/10 22:11:22 INFO DAGScheduler: ResultStage 15 (parquet at &lt;console&gt;:26) finished in 0.505 s
16/06/10 22:11:22 INFO DAGScheduler: Job 9 finished: parquet at &lt;console&gt;:26, took 5.054011 s
16/06/10 22:11:22 INFO ParquetRelation: Using default output committer for Parquet: org.apache.parquet.hadoop.ParquetOutputCommitter
16/06/10 22:11:22 INFO FileOutputCommitter: File Output Committer Algorithm version is 1
16/06/10 22:11:22 INFO DefaultWriterContainer: Using user defined output committer class org.apache.parquet.hadoop.ParquetOutputCommitter
16/06/10 22:11:22 INFO FileOutputCommitter: File Output Committer Algorithm version is 1
16/06/10 22:11:22 INFO SparkContext: Starting job: parquet at &lt;console&gt;:26
16/06/10 22:11:22 INFO DAGScheduler: Registering RDD 49 (parquet at &lt;console&gt;:26)
16/06/10 22:11:22 INFO DAGScheduler: Got job 10 (parquet at &lt;console&gt;:26) with 25 output partitions
16/06/10 22:11:22 INFO DAGScheduler: Final stage: ResultStage 18 (parquet at &lt;console&gt;:26)
16/06/10 22:11:22 INFO DAGScheduler: Parents of final stage: List(ShuffleMapStage 17)
16/06/10 22:11:22 INFO DAGScheduler: Missing parents: List(ShuffleMapStage 17)
16/06/10 22:11:22 INFO DAGScheduler: Submitting ShuffleMapStage 17 (MapPartitionsRDD[49] at parquet at &lt;console&gt;:26), which has no missing parents
16/06/10 22:11:22 INFO MemoryStore: Block broadcast_25 stored as values in memory (estimated size 17.4 KB, free 512.3 KB)
16/06/10 22:11:22 INFO MemoryStore: Block broadcast_25_piece0 stored as bytes in memory (estimated size 8.9 KB, free 521.2 KB)
16/06/10 22:11:22 INFO BlockManagerInfo: Added broadcast_25_piece0 in memory on 172.16.20.57:44944 (size: 8.9 KB, free: 517.3 MB)
16/06/10 22:11:22 INFO SparkContext: Created broadcast 25 from broadcast at DAGScheduler.scala:1006
16/06/10 22:11:22 INFO DAGScheduler: Submitting 200 missing tasks from ShuffleMapStage 17 (MapPartitionsRDD[49] at parquet at &lt;console&gt;:26)
16/06/10 22:11:22 INFO YarnScheduler: Adding task set 17.0 with 200 tasks
16/06/10 22:11:23 INFO TaskSetManager: Starting task 0.0 in stage 17.0 (TID 1125, slave-1, partition 0,NODE_LOCAL, 1988 bytes)
16/06/10 22:11:23 INFO TaskSetManager: Starting task 1.0 in stage 17.0 (TID 1126, slave-2, partition 1,NODE_LOCAL, 1988 bytes)
16/06/10 22:11:23 INFO TaskSetManager: Starting task 2.0 in stage 17.0 (TID 1127, slave-1, partition 2,NODE_LOCAL, 1988 bytes)
16/06/10 22:11:23 INFO TaskSetManager: Starting task 3.0 in stage 17.0 (TID 1128, slave-2, partition 3,NODE_LOCAL, 1988 bytes)
16/06/10 22:11:23 INFO TaskSetManager: Starting task 4.0 in stage 17.0 (TID 1129, slave-1, partition 4,NODE_LOCAL, 1988 bytes)
16/06/10 22:11:23 INFO TaskSetManager: Starting task 5.0 in stage 17.0 (TID 1130, slave-2, partition 5,NODE_LOCAL, 1988 bytes)
</code></pre>

<p><strong>Questions</strong></p>

<ol>
<li>Why two <code>Jobs</code>? What is the intention here by breaking a <code>DAG</code> into two <code>jobs</code>?</li>
<li><code>Job 10</code>'s <code>DAG</code> looks <strong>complete</strong> for the query execution. Is there anything specific <code>Job 9</code> is doing?</li>
<li>Why <code>Stage-17</code> is not Skipped? It looks like dummy <code>tasks</code> are created, do they have any purpose.</li>
<li><p>Later, I tried another rather simpler query. Unexpectedly, it was creating 3 <code>Jobs</code>.  </p>

<p>sqlContext.sql("select dpHour from so  order by dphour").write.parquet("/out2/")</p></li>
</ol>