title: Joining process with broadcast variable ends up endless spilling
body: <p>I am joining two RDDs from text files in standalone mode. One has 400 million (9 GB) rows, and the other has 4 million (110 KB). </p>
  
  <pre><code>3-grams  doc1           3-grams   doc2
  ion -    100772C111      ion -    200772C222  
  on  -    100772C111      gon -    200772C222  
   n  -    100772C111        n -    200772C222
  ... -    ....            ... -    .... 
  ion -    3332145654      on  -    58898874
  mju -    3332145654      mju -    58898874
  ... -    ....            ... -    ....
  </code></pre>
  
  <p>In each file, doc numbers (doc1 or doc2) appear one under the other. And as a result of join I would like to get a number of common 3-grams between the docs.e.g.</p>
  
  <pre><code>  (100772C111-200772C222,2) --&gt; There two common 3-grams which are 'ion' and  '  n'
  </code></pre>
  
  <p>The server on which I run my code has 128 GB RAM and 24 cores. I set my IntelliJ configurations - VM options part with  -Xmx64G</p>
  
  <p>Here is my code for this:</p>
  
  <pre><code>val conf = new SparkConf().setAppName("abdulhay").setMaster("local[4]").set("spark.shuffle.spill", "true")
        .set("spark.shuffle.memoryFraction", "0.6").set("spark.storage.memoryFraction", "0.4")
        .set("spark.executor.memory","40g")
        .set("spark.driver.memory","40g")
  
  val sc = new SparkContext(conf)
  
  val emp = sc.textFile("\\doc1.txt").map(line =&gt; (line.split("\t")(3),line.split("\t")(1))).distinct()
      val emp_new = sc.textFile("\\doc2.txt").map(line =&gt; (line.split("\t")(3),line.split("\t")(1))).distinct()
  
  val emp_newBC = sc.broadcast(emp_new.groupByKey.collectAsMap)
  
  val joined = emp.mapPartitions(iter =&gt; for {
        (k, v1) &lt;- iter
        v2 &lt;- emp_newBC.value.getOrElse(k, Iterable())
      } yield (s"$v1-$v2", 1))
  
  val olsun = joined.reduceByKey((a,b) =&gt; a+b)
  
  olsun.map(x =&gt; x._1 + "\t" + x._2).saveAsTextFile("...\\out.txt")
  </code></pre>
  
  <p>So as seen, during join process using broadcast variable my key values change. So it seems I need to repartition the joined values? And it is highly expensive. As a result, i ended up too much spilling issue, and it never ended. I think 128 GB memory must be sufficient. As far as I understood, when broadcast variable is used shuffling is being decreased significantly? So what is wrong with my application?</p>
  
  <p>Thanks in advance. </p>
  
  <p>EDIT:</p>
  
  <p>I have also tried spark's join function as below:</p>
  
  <pre><code>var joinRDD = emp.join(emp_new);
  
  val kkk = joinRDD.map(line =&gt; (line._2,1)).reduceByKey((a, b) =&gt; a + b)
  </code></pre>
  
  <p>again ending up too much spilling.</p>
  
  <p>EDIT2:</p>
  
  <pre><code>val conf = new SparkConf().setAppName("abdulhay").setMaster("local[12]").set("spark.shuffle.spill", "true")
        .set("spark.shuffle.memoryFraction", "0.4").set("spark.storage.memoryFraction", "0.6")
        .set("spark.executor.memory","50g")
        .set("spark.driver.memory","50g")
      val sc = new SparkContext(conf)
  
  val emp = sc.textFile("S:\\Staff_files\\Mehmet\\Projects\\SPARK - scala\\wos14.txt").map{line =&gt; val s = line.split("\t"); (s(5),s(0))}//.distinct()
      val emp_new = sc.textFile("S:\\Staff_files\\Mehmet\\Projects\\SPARK - scala\\fwo_word.txt").map{line =&gt; val s = line.split("\t"); (s(3),s(1))}//.distinct()
  
      val cog = emp_new.cogroup(emp)
  
  val skk =  cog.flatMap {
        case (key: String, (l1: Iterable[String], l2: Iterable[String])) =&gt;
          (l1.toSeq ++ l2.toSeq).combinations(2).map { case Seq(x, y) =&gt; if (x &lt; y) ((x, y),1) else ((y, x),1) }.toList
      }
  
      val com = skk.countByKey()
  </code></pre>