title: SPARK DataFrame: How to efficiently split dataframe for each group based on same column values
body: <p>I have a DataFrame generated as follows:</p>
  
  <pre><code>df.groupBy($"Hour", $"Category")
    .agg(sum($"value").alias("TotalValue"))
    .sort($"Hour".asc,$"TotalValue".desc))
  </code></pre>
  
  <p>The results look like:</p>
  
  <pre><code>+----+--------+----------+
  |Hour|Category|TotalValue|
  +----+--------+----------+
  |   0|   cat26|      30.9|
  |   0|   cat13|      22.1|
  |   0|   cat95|      19.6|
  |   0|  cat105|       1.3|
  |   1|   cat67|      28.5|
  |   1|    cat4|      26.8|
  |   1|   cat13|      12.6|
  |   1|   cat23|       5.3|
  |   2|   cat56|      39.6|
  |   2|   cat40|      29.7|
  |   2|  cat187|      27.9|
  |   2|   cat68|       9.8|
  |   3|    cat8|      35.6|
  | ...|    ....|      ....|
  +----+--------+----------+
  </code></pre>
  
  <p>I would like to make new dataframes based on every unique value of <code>col("Hour")</code> , i.e.</p>
  
  <ul>
  <li>for the group of Hour==0</li>
  <li>for the group of Hour==1 </li>
  <li>for the group of Hour==2
  and so on...</li>
  </ul>
  
  <p>So the desired output would be:</p>
  
  <pre><code>df0 as:
  
  +----+--------+----------+
  |Hour|Category|TotalValue|
  +----+--------+----------+
  |   0|   cat26|      30.9|
  |   0|   cat13|      22.1|
  |   0|   cat95|      19.6|
  |   0|  cat105|       1.3|
  +----+--------+----------+
  
  df1 as:
  +----+--------+----------+
  |Hour|Category|TotalValue|
  +----+--------+----------+
  |   1|   cat67|      28.5|
  |   1|    cat4|      26.8|
  |   1|   cat13|      12.6|
  |   1|   cat23|       5.3|
  +----+--------+----------+
  </code></pre>
  
  <p>and similarly,</p>
  
  <pre><code>df2 as:
  
  +----+--------+----------+
  |Hour|Category|TotalValue|
  +----+--------+----------+
  |   2|   cat56|      39.6|
  |   2|   cat40|      29.7|
  |   2|  cat187|      27.9|
  |   2|   cat68|       9.8|
  +----+--------+----------+
  </code></pre>
  
  <p>Any help is highly appreciated.</p>
  
  <p><strong><em>EDIT 1:</em></strong></p>
  
  <p>What I have tried:</p>
  
  <pre><code>df.foreach(
    row =&gt; splitHour(row)
    )
  
  def splitHour(row: Row) ={
      val Hour=row.getAs[Long]("Hour")
  
      val HourDF= sparkSession.createDataFrame(List((s"$Hour",1)))
  
      val hdf=HourDF.withColumnRenamed("_1","Hour_unique").drop("_2")
  
      val mydf: DataFrame =df.join(hdf,df("Hour")===hdf("Hour_unique"))
  
      mydf.write.mode("overwrite").parquet(s"/home/dev/shaishave/etc/myparquet/$Hour/")
    }
  </code></pre>
  
  <p>PROBLEM WITH THIS STRATEGY:</p>
  
  <p>It took 8 hours when it was run on a dataframe <code>df</code> which had over 1 million rows and spark job was given around 10 GB RAM on single node. So, <code>join</code> is turning out to be highly in-efficient. </p>
  
  <p>Caveat: I have to write each dataframe <code>mydf</code> as parquet which has nested schema that is required to be maintained (not flattened).</p>