Spark is taking too much time and creating thousands of jobs for some tasks
<p>Machine Config :</p>

<ul>
<li>RAM: 16 gb</li>
<li>Processor: 4 cores(Xeon E3 3.3 GHz) </li>
</ul>

<p>Problem:</p>

<ul>
<li>Time Consuming : Taking more than 18 minutes </li>
</ul>

<p>Case Scenario :</p>

<ul>
<li><blockquote>
  <p>Spark Mode: Local</p>
</blockquote></li>
<li><p>Database: Using Cassandra 2.1.12 </p></li>
</ul>

<p>I am fetching 3 tables into dataframes , which is having less than 10 rows. yes, less than 10 (ten). 
After fetching it into dataframes I performing joins,count,show and collect operation many times. When I execute my program Spark is creating 40404 jobs 4 times. it indicates that count requires to perform those jobs. I am using count 4-5 times in program. After waiting for more than 18 minutes(approx 18.5 to 20) it gives me expected output.</p>

<ul>
<li>why Spark is creating that much of jobs?</li>
<li>is it obvious ('ok') to take this much time (18 minutes) to execute this number of jobs(40404 * 4 approx)?</li>
</ul>

<p>Thanks in advance.</p>

<p>Sample code 1:</p>



<pre class="lang-scala prettyprint-override"><code>def getGroups(id: Array[String], level: Int): DataFrame = {
    var lvl = level
    if (level &gt;= 0) {
      for (iterated_id &lt;- id) {
        val single_level_group = supportive_df.filter("id = '" + iterated_id + "' and level = " + level).select("family_id")
        //single_level_group.show()
        intermediate_df = intermediate_df.unionAll(single_level_group)
        //println("for loop portion...")
      }
      final_df = final_df.unionAll(intermediate_df)
      lvl -= 1
      val user_id_param = intermediate_df.collect().map { row =&gt; row.getString(0) }
      intermediate_df = empty_df
      //println("new method...if portion...")
      getGroups(user_id_param, lvl)
    } else {
      //println("new method...")
      final_df.distinct()
    }
  }
</code></pre>

<p>Sample code 2:</p>

<pre class="lang-scala prettyprint-override"><code> setGetGroupsVars("u_id", user_id.toString(), sa_user_df)
  var user_belong_groups: DataFrame = empty_df
  val user_array = Array[String](user_id.toString())

  val user_levels = sa_user_df.filter("id = '" + user_id + "'").select("level").distinct().collect().map { x =&gt; x.getInt(0) }

  println(user_levels.length+"...rapak")
  println(user_id.toString())
  for (u_lvl &lt;- user_levels) {
    val x1 = getGroups(user_array, u_lvl)
    x1.show()
    empty_df.show()
    user_belong_groups.show()
    user_belong_groups = user_belong_groups.unionAll(x1)
    x1.show()
  }
  setGetGroupsVars("obj_id", obj_id.toString(), obj_type_specific_df)
  var obj_belong_groups: DataFrame = empty_df
  val obj_array = Array[String](obj_id.toString())
  val obj_levels = obj_type_specific_df.filter("id = '" + obj_id + "'").select("level").distinct().collect().map { x =&gt; x.getInt(0) }
  println(obj_levels.length)
  for (ob_lvl &lt;- obj_levels) {
    obj_belong_groups = obj_belong_groups.unionAll(getGroups(obj_array, ob_lvl))
  }
  user_belong_groups = user_belong_groups.distinct()
  obj_belong_groups = obj_belong_groups.distinct()
  var user_obj_joined_df = user_belong_groups.join(obj_belong_groups)
  user_obj_joined_df.show()

  println("vbgdivsivbfb")
  var user_obj_access_df = user_obj_joined_df
    .join(sa_other_access_df, user_obj_joined_df("u_id") === sa_other_access_df("user_id")
      &amp;&amp; user_obj_joined_df("obj_id") === sa_other_access_df("object_id"))
  user_obj_access_df.show()
  println("KDDD..")

  val user_obj_access_cond1 = user_obj_access_df.filter("u_id = '" + user_id + "' and obj_id != '" + obj_id + "'")
  if (user_obj_access_cond1.count() == 0) {
    val user_obj_access_cond2 = user_obj_access_df.filter("u_id != '" + user_id + "' and obj_id = '" + obj_id + "'")
    if (user_obj_access_cond2.count() == 0) {
      val user_obj_access_cond3 = user_obj_access_df.filter("u_id != '" + user_id + "' and obj_id != '" + obj_id + "'")
      if (user_obj_access_cond3.count() == 0) {
        default_df
      } else {
        val result_ugrp_to_objgrp = user_obj_access_cond3.select("permission").agg(max("permission"))
        println("cond4")
        result_ugrp_to_objgrp
      }
    } else {
      val result_ugrp_to_ob = user_obj_access_cond2.select("permission")
      println("cond3")
      result_ugrp_to_ob
    }
  } else {
    val result_u_to_obgrp = user_obj_access_cond1.select("permission")
    println("cond2")
    result_u_to_obgrp
  }
} else {
  println("cond1")
  individual_access
}
</code></pre>

<p>These two are major code block in my prog where the execution is taking too longer. It generally takes much time at show or count operation.</p>