title: Apache Spark - shuffle writes more data than the size of the input data
body: <p>I use Spark 2.1 in local mode and I'm running this simple application.</p>
  
  <pre><code>val N = 10 &lt;&lt; 20
  
  sparkSession.conf.set("spark.sql.shuffle.partitions", "5")
  sparkSession.conf.set("spark.sql.autoBroadcastJoinThreshold", (N + 1).toString)
  sparkSession.conf.set("spark.sql.join.preferSortMergeJoin", "false")
  
  val df1 = sparkSession.range(N).selectExpr(s"id as k1")
  val df2 = sparkSession.range(N / 5).selectExpr(s"id * 3 as k2")
  
  df1.join(df2, col("k1") === col("k2")).count()
  </code></pre>
  
  <p>Here, the range(N) creates a dataset of <em>Long</em> (with unique values), so I assume that the size of </p>
  
  <blockquote>
    <ul>
    <li>df1 = N * 8 bytes ~ 80MB</li>
    <li>df2 = N / 5 * 8 bytes ~ 16MB</li>
    </ul>
  </blockquote>
  
  <p>Ok now let's take df1 as an example.
  <strong>df1 consists of 8 partitions</strong> and <strong>shuffledRDDs of 5</strong>, so I assume that</p>
  
  <blockquote>
    <ul>
    <li># of mappers (M) = 8</li>
    <li># of reducers (R) = 5</li>
    </ul>
  </blockquote>
  
  <p>As the # of partitions is low, Spark will use the Hash Shuffle which will create <strong>M * R files</strong> in the disk but I haven't understood if every file has all the data, thus <strong>each_file_size = data_size</strong> resulting to <strong>M * R * data_size</strong> files or <strong>all_files = data_size</strong>.</p>
  
  <p>However when executing this app, shuffle write of <strong>df1 = 160MB</strong> which doesn't match either of the above cases.</p>
  
  <p><a href="https://i.stack.imgur.com/SUp1T.png" rel="nofollow noreferrer">Spark UI</a></p>
  
  <p>What am I missing here? Why has the shuffle write data doubled in size?</p>