title: Why does orderBy() modify the result of aggregation on a DataFrame with PySpark?
body: <p>I have to use PySpark to find <strong>the most often sold products together</strong> from data stored on a dataframe with name <code>sales_NY</code> and whose sample is:</p>
  <pre><code>+-------+--------------------+--------+------+-------------------+--------------------+-------------+-----+-----+----+
  |OrderID|             Product|Quantity| Price|          OrderDate|        StoreAddress|         City|State|Month|Hour|
  +-------+--------------------+--------+------+-------------------+--------------------+-------------+-----+-----+----+
  | 295665|  Macbook Pro Laptop|       1|1700.0|2019-12-30 00:01:00|136 Church St, Ne...|New York City|   NY|   12|   0|
  | 295666|  LG Washing Machine|       1| 600.0|2019-12-29 07:03:00|562 2nd St, New Y...|New York City|   NY|   12|   7|
  | 295667|USB-C Charging Cable|       1| 11.95|2019-12-12 18:21:00|277 Main St, New ...|New York City|   NY|   12|  18|
  | 295670|AA Batteries (4-p...|       1|  3.84|2019-12-31 22:58:00|200 Jefferson St,...|New York City|   NY|   12|  22|
  | 295698|     Vareebadd Phone|       1| 400.0|2019-12-13 14:32:00|175 1st St, New Y...|New York City|   NY|   12|  14|
  | 295698|USB-C Charging Cable|       2| 11.95|2019-12-13 14:32:00|175 1st St, New Y...|New York City|   NY|   12|  14|
  | 295700|Bose SoundSport H...|       1| 99.99|2019-12-25 19:02:00|363 Hickory St, N...|New York City|   NY|   12|  19|
  | 295704|    Wired Headphones|       1| 11.99|2019-12-12 00:20:00|457 8th St, New Y...|New York City|   NY|   12|   0|
  | 295705|    Wired Headphones|       1| 11.99|2019-12-25 10:41:00|133 Jackson St, N...|New York City|   NY|   12|  10|
  | 295712|  Macbook Pro Laptop|       1|1700.0|2019-12-10 20:02:00|331 Madison St, N...|New York City|   NY|   12|  20|
  | 295713|Bose SoundSport H...|       1| 99.99|2019-12-24 07:55:00|490 Spruce St, Ne...|New York City|   NY|   12|   7|
  | 295720|AA Batteries (4-p...|       1|  3.84|2019-12-17 22:52:00|298 Ridge St, New...|New York City|   NY|   12|  22|
  | 295728|    27in FHD Monitor|       1|149.99|2019-12-21 19:21:00|366 Washington St...|New York City|   NY|   12|  19|
  | 295735|              iPhone|       1| 700.0|2019-12-22 18:25:00|374 Lincoln St, N...|New York City|   NY|   12|  18|
  | 295735|Apple Airpods Hea...|       1| 150.0|2019-12-22 18:25:00|374 Lincoln St, N...|New York City|   NY|   12|  18|
  | 295735|    Wired Headphones|       1| 11.99|2019-12-22 18:25:00|374 Lincoln St, N...|New York City|   NY|   12|  18|
  | 295740|USB-C Charging Cable|       1| 11.95|2019-12-01 20:36:00|102 Cedar St, New...|New York City|   NY|   12|  20|
  | 295742|Apple Airpods Hea...|       1| 150.0|2019-12-09 23:45:00|368 Sunset St, Ne...|New York City|   NY|   12|  23|
  | 295743|USB-C Charging Cable|       1| 11.95|2019-12-03 11:52:00|346 South St, New...|New York City|   NY|   12|  11|
  | 295745|       Flatscreen TV|       1| 300.0|2019-12-24 10:38:00|124 Lakeview St, ...|New York City|   NY|   12|  10|
  +-------+--------------------+--------+------+-------------------+--------------------+-------------+-----+-----+----+
  </code></pre>
  <p>To find <strong>the most often sold products together</strong> I'm using a first part of the code where I load the data into memory (sorry for the lack of reproducibility, but the size of the whole data is quite large to include it here):</p>
  <p><strong>Common part:</strong></p>
  <pre><code>from pyspark.sql import SparkSession
  from pyspark.sql.functions import col, collect_list, size, hour
  
  spark = (SparkSession.builder.appName('SalesAnalytics').getOrCreate())
  
  ### This is the local path to my data:
  file_path = './data/output/sales/ReportYear=2019'
  sales_raw_df = (spark.read.format('parquet')
                   .option('header', 'True')
                   .option('inferSchema', 'True')
                   .load(file_path))
  sales_raw_df = sales_raw_df.withColumn('Hour', hour(sales_raw_df.OrderDate))
  
  sales_NY = (sales_raw_df.where(col('State') == 'NY'))
  </code></pre>
  <p>Now I may follow 2 different versions of the solution, which I believed being fully equivalent, but the outputs are slightly different. The versions differ in that the second one adds an intermediate step using <code>orderBy('OrderID', 'State')</code>.</p>
  <p><strong>Version 1:</strong></p>
  <pre><code>sales_q4_df = (sales_NY.groupBy('OrderID', 'State')
                         .agg(collect_list('Product').alias('ProductList')))
  sales_q4_df = (sales_q4_df.withColumn('ProductListSize', size('ProductList')))
  
  ### discards the orders with a single product (the OrderID just appears once)
  sales_q4_df = sales_q4_df.filter(col('ProductListSize') &gt; 1).orderBy('ProductList', ascending=True)
  most_prods_together = sales_q4_df.groupBy('ProductList').count().orderBy('count', ascending=False).show(10, False)
  </code></pre>
  <p><strong>Output 1:</strong></p>
  <pre><code>+------------------------------------------------------+-----+
  |ProductList                                           |count|
  +------------------------------------------------------+-----+
  |[iPhone, Lightning Charging Cable]                    |126  |
  |[Google Phone, USB-C Charging Cable]                  |124  |
  |[Google Phone, Wired Headphones]                      |52   |
  |[Vareebadd Phone, USB-C Charging Cable]               |49   |
  |[iPhone, Wired Headphones]                            |46   |
  |[iPhone, Apple Airpods Headphones]                    |43   |
  |[Google Phone, Bose SoundSport Headphones]            |23   |
  |[Vareebadd Phone, Wired Headphones]                   |17   |
  |[Apple Airpods Headphones, Wired Headphones]          |12   |
  |[Google Phone, USB-C Charging Cable, Wired Headphones]|11   |
  +------------------------------------------------------+-----+
  </code></pre>
  <p><strong>Version 2:</strong></p>
  <pre><code>sales_q4_df = (sales_NY.orderBy('OrderID', 'Product')
                         .groupBy('OrderID', 'State')
                         .agg(collect_list('Product').alias('ProductList')))
  sales_q4_df = (sales_q4_df.withColumn('ProductListSize', size('ProductList')))
  
  ### discards the orders with a single product (the OrderID just appears once)
  sales_q4_df = sales_q4_df.filter(col('ProductListSize') &gt; 1).orderBy('ProductList', ascending=True)
  most_prods_together = sales_q4_df.groupBy('ProductList').count().orderBy('count', ascending=False).show(10, False)
  </code></pre>
  <p><strong>Output 2:</strong></p>
  <pre><code>+-------------------------------------------------+-----+
  |ProductList                                      |count|
  +-------------------------------------------------+-----+
  |[Google Phone, USB-C Charging Cable]             |127  |
  |[Lightning Charging Cable, iPhone]               |126  |
  |[Google Phone, Wired Headphones]                 |53   |
  |[USB-C Charging Cable, Vareebadd Phone]          |50   |
  |[Wired Headphones, iPhone]                       |46   |
  |[Apple Airpods Headphones, iPhone]               |45   |
  |[Bose SoundSport Headphones, Google Phone]       |24   |
  |[Apple Airpods Headphones, Wired Headphones]     |19   |
  |[Vareebadd Phone, Wired Headphones]              |17   |
  |[AA Batteries (4-pack), Lightning Charging Cable]|16   |
  +-------------------------------------------------+-----+
  </code></pre>
  <p>Can anyone explain me why the results are different? Is this a bug in PySpark?</p>
  <p>I am working on Notebooks with jupyterlab v.3.4.2, PySpark v.3.0.1 and Java v.15.</p>
  <p>PD: I must add that I tried as well the method <code>sort()</code> (more efficient than <code>orderBy()</code> because of using several partitions), but the result was the same.</p>