title: spark- group all elements of array column into batches using scala
body: <p>I have a dataframe with a column of array type where the number of elements differs between rows as seen in GPS_Array_Size of the below Input Dataframe.</p>
  <p>I need to post an http request to an external service with 5000 elements at a time by also embedding the &quot;requestid&quot; field into each tuple as shown in the Output Dataframe.</p>
  <p><strong>Input Dataframe schema:-</strong></p>
  <pre><code>root
   |-- requestid: string (nullable = true)
   |-- GPS_Array: array (nullable = true)
   |    |-- element: struct (containsNull = true)
   |    |    |-- timestamp: double (nullable = true)
   |    |    |-- GPSLatitude: double (nullable = true)
   |    |    |-- GPSLongitude: double (nullable = true)
   |-- GPS_Array_Size: long (nullable = true)
  </code></pre>
  <p><strong>Input DataFrame:-</strong></p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th style="text-align: left;">requestid</th>
  <th style="text-align: center;">GPS_Array</th>
  <th style="text-align: center;">GPS_Array_Size</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td style="text-align: left;">aaa</td>
  <td style="text-align: center;">[{&quot;timestamp&quot;:a1,&quot;GPSLatitude&quot;:a1,&quot;GPSLongitude&quot;:a1},{&quot;timestamp&quot;:a2,&quot;GPSLatitude&quot;:a2,&quot;GPSLongitude&quot;:a2},...{&quot;timestamp&quot;:a6431,&quot;GPSLatitude&quot;:a6431,&quot;GPSLongitude&quot;:a6431}]</td>
  <td style="text-align: center;">6431</td>
  </tr>
  <tr>
  <td style="text-align: left;">bbb</td>
  <td style="text-align: center;">[{&quot;timestamp&quot;:b1,&quot;GPSLatitude&quot;:b1,&quot;GPSLongitude&quot;:b1},{&quot;timestamp&quot;:b2,&quot;GPSLatitude&quot;:b2,&quot;GPSLongitude&quot;:b2},...{&quot;timestamp&quot;:b11876,&quot;GPSLatitude&quot;:b11876,&quot;GPSLongitude&quot;:b11876}]</td>
  <td style="text-align: center;">11876</td>
  </tr>
  <tr>
  <td style="text-align: left;">ccc</td>
  <td style="text-align: center;">[{&quot;timestamp&quot;:c1,&quot;GPSLatitude&quot;:c1,&quot;GPSLongitude&quot;:c1},{&quot;timestamp&quot;:c2,&quot;GPSLatitude&quot;:c2,&quot;GPSLongitude&quot;:c2},...{&quot;timestamp&quot;:c763,&quot;GPSLatitude&quot;:c763,&quot;GPSLongitude&quot;:c763}]</td>
  <td style="text-align: center;">763</td>
  </tr>
  <tr>
  <td style="text-align: left;">ddd</td>
  <td style="text-align: center;">[{&quot;timestamp&quot;:d1,&quot;GPSLatitude&quot;:d1,&quot;GPSLongitude&quot;:d1},{&quot;timestamp&quot;:d2,&quot;GPSLatitude&quot;:d2,&quot;GPSLongitude&quot;:d2},...{&quot;timestamp&quot;:d5187,&quot;GPSLatitude&quot;:d5187,&quot;GPSLongitude&quot;:d5187}]</td>
  <td style="text-align: center;">5187</td>
  </tr>
  <tr>
  <td style="text-align: left;">eee</td>
  <td style="text-align: center;">[{&quot;timestamp&quot;:e1,&quot;GPSLatitude&quot;:e1,&quot;GPSLongitude&quot;:e1},{&quot;timestamp&quot;:e2,&quot;GPSLatitude&quot;:e2,&quot;GPSLongitude&quot;:e2},...{&quot;timestamp&quot;:e1023,&quot;GPSLatitude&quot;:e1023,&quot;GPSLongitude&quot;:e1023}]</td>
  <td style="text-align: center;">1023</td>
  </tr>
  </tbody>
  </table>
  </div>
  <p><strong>Output Dataframe schema:-</strong></p>
  <pre><code>root
   |-- New_GPS_Array: array (nullable = true)
   |    |-- element: struct (containsNull = true)
   |    |    |-- requestid: string (nullable = true)
   |    |    |-- timestamp: double (nullable = true)
   |    |    |-- GPSLatitude: double (nullable = true)
   |    |    |-- GPSLongitude: double (nullable = true)
   |-- New_GPS_Array_Size: long (nullable = true)
  </code></pre>
  <p><strong>Output DataFrame:-</strong></p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th style="text-align: left;">New_GPS_Array</th>
  <th style="text-align: center;">New_GPS_Array_Size</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td style="text-align: left;">[{&quot;requestid&quot;:&quot;aaa_1&quot;,&quot;timestamp&quot;:a1,&quot;GPSLatitude&quot;:a1,&quot;GPSLongitude&quot;:a1},{&quot;requestid&quot;:&quot;aaa_1&quot;,&quot;timestamp&quot;:a2,&quot;GPSLatitude&quot;:a2,&quot;GPSLongitude&quot;:a2},...{&quot;requestid&quot;:&quot;aaa_1&quot;,&quot;timestamp&quot;:a5000,&quot;GPSLatitude&quot;:a5000,&quot;GPSLongitude&quot;:a5000}]</td>
  <td style="text-align: center;">5000</td>
  </tr>
  <tr>
  <td style="text-align: left;">[{&quot;requestid&quot;:&quot;aaa_2&quot;,&quot;timestamp&quot;:a5001,&quot;GPSLatitude&quot;:a5001,&quot;GPSLongitude&quot;:a5001},{&quot;requestid&quot;:&quot;aaa_2&quot;,&quot;timestamp&quot;:a5002,&quot;GPSLatitude&quot;:a5002,&quot;GPSLongitude&quot;:a5002},...{&quot;requestid&quot;:&quot;aaa_2&quot;,&quot;timestamp&quot;:a6431,&quot;GPSLatitude&quot;:a6431,&quot;GPSLongitude&quot;:a6431},{&quot;requestid&quot;:&quot;bbb_1&quot;,&quot;timestamp&quot;:b1,&quot;GPSLatitude&quot;:b1,&quot;GPSLongitude&quot;:b1},...{&quot;requestid&quot;:&quot;bbb_1&quot;,&quot;timestamp&quot;:b3569,&quot;GPSLatitude&quot;:b3569,&quot;GPSLongitude&quot;:b3569} ]</td>
  <td style="text-align: center;">5000</td>
  </tr>
  <tr>
  <td style="text-align: left;">[{&quot;requestid&quot;:&quot;bbb_2&quot;,&quot;timestamp&quot;:b3570,&quot;GPSLatitude&quot;:b3570,&quot;GPSLongitude&quot;:b3570},{&quot;requestid&quot;:&quot;bbb_2&quot;,&quot;timestamp&quot;:b3571,&quot;GPSLatitude&quot;:b3571,&quot;GPSLongitude&quot;:b3571},...{&quot;requestid&quot;:&quot;bbb_2&quot;,&quot;timestamp&quot;:b8569,&quot;GPSLatitude&quot;:b8569,&quot;GPSLongitude&quot;:b8569} ]</td>
  <td style="text-align: center;">5000</td>
  </tr>
  <tr>
  <td style="text-align: left;">[{&quot;requestid&quot;:&quot;bbb_3&quot;,&quot;timestamp&quot;:b8570,&quot;GPSLatitude&quot;:b8570,&quot;GPSLongitude&quot;:b8570},{&quot;requestid&quot;:&quot;bbb_3&quot;,&quot;timestamp&quot;:b8571,&quot;GPSLatitude&quot;:b8571,&quot;GPSLongitude&quot;:b8571},...{&quot;requestid&quot;:&quot;bbb_3&quot;,&quot;timestamp&quot;:b11876,&quot;GPSLatitude&quot;:b11876,&quot;GPSLongitude&quot;:b11876},{&quot;requestid&quot;:&quot;ccc_1&quot;,&quot;timestamp&quot;:c1,&quot;GPSLatitude&quot;:c1,&quot;GPSLongitude&quot;:c1},...{&quot;requestid&quot;:&quot;ccc_1&quot;,&quot;timestamp&quot;:c763,&quot;GPSLatitude&quot;:c763,&quot;GPSLongitude&quot;:c763},{&quot;requestid&quot;:&quot;ddd_1&quot;,&quot;timestamp&quot;:d1,&quot;GPSLatitude&quot;:d1,&quot;GPSLongitude&quot;:d1},...{&quot;requestid&quot;:&quot;ddd_1&quot;,&quot;timestamp&quot;:d930,&quot;GPSLatitude&quot;:d930,&quot;GPSLongitude&quot;:d930} ]</td>
  <td style="text-align: center;">5000</td>
  </tr>
  <tr>
  <td style="text-align: left;">[{&quot;requestid&quot;:&quot;ddd_2&quot;,&quot;timestamp&quot;:d931,&quot;GPSLatitude&quot;:d931,&quot;GPSLongitude&quot;:d931},{&quot;requestid&quot;:&quot;ddd_2&quot;,&quot;timestamp&quot;:d932,&quot;GPSLatitude&quot;:d932,&quot;GPSLongitude&quot;:d932},...{&quot;requestid&quot;:&quot;ddd_2&quot;,&quot;timestamp&quot;:d5187,&quot;GPSLatitude&quot;:d5187,&quot;GPSLongitude&quot;:d5187},{&quot;requestid&quot;:&quot;eee_1&quot;,&quot;timestamp&quot;:e1,&quot;GPSLatitude&quot;:e1,&quot;GPSLongitude&quot;:e1},...{&quot;requestid&quot;:&quot;eee_1&quot;,&quot;timestamp&quot;:e743,&quot;GPSLatitude&quot;:e743,&quot;GPSLongitude&quot;:e743} ]</td>
  <td style="text-align: center;">5000</td>
  </tr>
  <tr>
  <td style="text-align: left;">[{&quot;requestid&quot;:&quot;eee_2&quot;,&quot;timestamp&quot;:e744,&quot;GPSLatitude&quot;:e744,&quot;GPSLongitude&quot;:e744},{&quot;requestid&quot;:&quot;eee_2&quot;,&quot;timestamp&quot;:e745,&quot;GPSLatitude&quot;:e745,&quot;GPSLongitude&quot;:e745},...{&quot;requestid&quot;:&quot;eee_2&quot;,&quot;timestamp&quot;:e1023,&quot;GPSLatitude&quot;:e1023,&quot;GPSLongitude&quot;:e1023}]</td>
  <td style="text-align: center;">280</td>
  </tr>
  </tbody>
  </table>
  </div>
  <p>As shown requestid needs to change for every new batch.</p>
  <p>My Solution was to use the explode function to create a row for every element in the GPS_Array and then assign an index based on row number. So if I have 700 million rows after exploding then 1 to 700,000,000. The index is divided by 5000 to get the quotient and then the quotients are grouped.</p>
  <p>​Row numbers 0 to 4999 will have quotient 0, row numbers 5000 to 9999 will have quotient 1 and row numbers 10000 to 14999 will have quotient 2 ..etc</p>
  <p>explodedDF.withColumn(&quot;index&quot;, row_number().over(Window.orderBy(monotonically_increasing_id()))).withColumn(&quot;id&quot;, floor($&quot;id&quot; / 5000).groupBy(&quot;id&quot;).agg(collect_list($&quot;GPS_Array&quot;))</p>
  <p>I believe the part that does the indexing brings all the records to 1 spark worker node since it has to auto increment over each record.</p>
  <p>Can batching into 5000 elements per request be done more efficiently?</p>
  <p>Version : (spark 2.45 and scala 2.11)</p>