Is groupby a computationally expensive operation in pyspark
https://stackoverflow.com/questions/44734088/improve-groupby-operation-in-spark-1-5-2
Improve groupby operation in Spark 1.5.2
2017-06-28
Thejas B Thejas B
<div><div>
 
<p>We are facing poor performance using Spark.</p>
<p>I have 2 specific questions:</p>
<ol>
<li>When debugging we noticed that a few of the groupby operations done on Rdd are taking more time </li>
<li>Also a few of the stages are appearing twice, some finishing very quickly, some taking more time </li>
</ol>
<p>Here is a screenshot of .</p>
<p>Currently running locally, having shuffle partitions set to 2 and number of partitions set to 5, data is around 1,00,000 records.</p>
<p>Speaking of groupby operation, we are grouping a dataframe (which is a result of several joins) based on two columns, and then applying a function to get some result.</p>
<pre><code> val groupedRows = rows.rdd.groupBy(row =&gt; (
 row.getAs[Long](Column1),
 row.getAs[Int](Column2)
 ))
 val rdd = groupedRows.values.map(Criteria)
</code></pre>
<p>Where Criteria is some function acted on the grouped resultant rows. Can we optimize this group by in any way?</p>
<p>Here is a screenshot of the .</p>
 </div>||||I|||| Stack Overflow
 1. About
 2. Products
 3. For Teams
 1. Stack Overflow Public questions & answers
 2. Stack Overflow for Teams Where developers & technologists share private knowledge with coworkers
 3. Talent Build your employer brand
 4. Advertising Reach developers & technologists worldwide
 5. About the company
 Loading…

 1. current community

 + Stack Overflow
 help chat
 + Meta Stack Overflow

 your communities

 Sign up or log in to customize your list.

 more stack exchange communities

 company blog
 2. 
 3. Log in
 4. Sign up
 1. Home
 2. 
 1. Public
 2. Questions
 3. Tags
 4. Users
 5. Companies
 6. Collectives
 7. Explore Collectives
 3. 
 1. Teams
 Stack Overflow for Teams – Start collaborating and sharing organizational knowledge. Create a free Team Why Teams?
 2. Teams
 3. Create free Team

 Collectives™ on Stack Overflow

 Find centralized, trusted content and collaborate around the technologies you use most.

 Learn more about Collectives

 Teams

 Q&A for work

 Connect and share knowledge within a single location that is structured and easy to search.

 Learn more about Teams

 Improve groupby operation in Spark 1.5.2

 Ask Question
 Asked 5 years, 7 months ago
 Modified 5 years, 7 months ago
 Viewed 2k times
 0

 We are facing poor performance using Spark.

 I have 2 specific questions:

 1. When debugging we noticed that a few of the groupby operations done on Rdd are taking more time
 2. Also a few of the stages are appearing twice, some finishing very quickly, some taking more time

 Here is a screenshot of .

 Currently running locally, having shuffle partitions set to 2 and number of partitions set to 5, data is around 1,00,000 records.

 Speaking of groupby operation, we are grouping a dataframe (which is a result of several joins) based on two columns, and then applying a function to get some result.

 val groupedRows = rows.rdd.groupBy(row => (
 row.getAs[Long](Column1),
 row.getAs[Int](Column2)
 ))
 val rdd = groupedRows.values.map(Criteria)
 

 Where Criteria is some function acted on the grouped resultant rows. Can we optimize this group by in any way?

 Here is a screenshot of the .

 * scala
 * apache-spark
 * group-by
 * apache-spark-sql
 Share
 Improve this question
 Follow
 edited Jun 28, 2017 at 2:17
 niton
 8,404 20 20 gold badges 31 31 silver badges 51 51 bronze badges
 asked Jun 24, 2017 at 7:38
 Thejas B Thejas B
 59 1 1 gold badge 1 1 silver badge 9 9 bronze badges
 Add a comment |

 2 Answers

 Sorted by: Reset to default
 Highest score (default) Trending (recent votes count more) Date modified (newest first) Date created (oldest first)
 0

 I would suggest you not to convert the existing dataframe to rdd and do the complex process you are performing.

 If you want to perform Criteria function on two columns (Column1 and Column2), you can do this directly on dataframe. Moreover, if your Criteria can be reduced to combination of inbuilt functions then it would be great. But you can always use udf functions for custom rules.

 What I would suggest you to do is groupBy on the dataframe and apply aggregation functions

 rows.groupBy("Column1", "Column2").agg(Criteria function)
 

 You can use Window functions if you want multiple rows from the grouped dataframe. more info here

 Share
 Improve this answer
 Follow
 answered Jun 24, 2017 at 8:02
 Ramesh Maharjan Ramesh Maharjan
 40.4k 6 6 gold badges 68 68 silver badges 94 94 bronze badges
 2
 * I dont think Spark 1.5.2 agg operation take in a function, it just takes in inbuilt sum , avg etc functions. Criteria is something we run each record for the group and select a record. After grouping we get RDD[((Long, Int), Iterable[Row])] and for each (key, key) we get a seq of rows, which we give it to Criteria
 – Thejas B
 Jun 24, 2017 at 8:33
 * For that you should have a look at udaf aggregations. UDAF seems to be appropriate solution
 – Ramesh Maharjan
 Jun 24, 2017 at 8:39
 Add a comment |
 0

 .groupBy is known to be not the most efficient approach:

 Note: This operation may be very expensive. If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using PairRDDFunctions.aggregateByKey or PairRDDFunctions.reduceByKey will provide much better performance.

 Sometimes it is better to use .reduceByKey or .aggregateByKey, as explained here:

 While both of these functions will produce the correct answer, the reduceByKey example works much better on a large dataset. That's because Spark knows it can combine output with a common key on each partition before shuffling the data.

 Why .reduceByKey, .aggregateByKey work faster than .groupBy? Because part of the aggregation happens during map phase and less data is shuffled around worker nodes during reduce phase. Here is a good explanation on how does aggregateByKey work.

 Share
 Improve this answer
 Follow
 answered Jun 25, 2017 at 10:58
 Nikolay Vasiliev Nikolay Vasiliev
 5,466 21 21 silver badges 31 31 bronze badges
 4
 * Thanks, replaced the groupby with aggregateByKey, seems much better now. But I still see duplicate stages getting executed again. You have any idea about it? I searched a lot in google but can't seem to get the answer.
 – Thejas B
 Jun 25, 2017 at 16:22
 * I would advise to check the DAG of the execution and the Event timeline. From the image you posted it's hard to say. If you define RDD operations in a sub-function and call it twice, it may happen that some RDD actions appear in the timeline twice for different data.
 – Nikolay Vasiliev
 Jun 25, 2017 at 22:34
 * Hi Nikolay, I've added a part of DAG graph, since I can't upload more than 2 images.
 – Thejas B
 Jun 27, 2017 at 5:23
 * Thanks. Looks like you are running Spark 2, since Tungsten engine is new to Spark 2 ( databricks.com/blog/2016/05/23/… ). I assume what you call "appear twice" is about jobs 8, 9 and 10. Since I don't see the code, I assume that either there is a loop that generates those stages (=> they appear several times) or Tungsten engine optimizes it for you (and generates code somehow). I don't think I can provide more help here :) Good luck and have fun.
 – Nikolay Vasiliev
 Jun 27, 2017 at 21:23
 Add a comment |

 Your Answer

 Thanks for contributing an answer to Stack Overflow!

 * Please be sure to answer the question. Provide details and share your research!

 But avoid …

 * Asking for help, clarification, or responding to other answers.
 * Making statements based on opinion; back them up with references or personal experience.

 To learn more, see our tips on writing great answers.

 Sign up or log in

 Sign up using Google
 Sign up using Facebook
 Sign up using Email and Password
 Submit

 Post as a guest

 Name
 Email

 Required, but never shown

 Post as a guest

 Name
 Email

 Required, but never shown

 Post Your Answer Discard

 By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

 Not the answer you're looking for? Browse other questions tagged
 * scala
 * apache-spark
 * group-by
 * apache-spark-sql
 or ask your own question.

 * The Overflow Blog
 * The nature of simulating nature: Q&A with IBM quantum computing research
 * The AI that writes music from text (Ep. 535)
 * Featured on Meta
 * Accessibility Update: Colors
 * Should we burninate the [balance] tag?
 * Collectives: The next iteration
 * Temporary policy: ChatGPT is banned
 Visit chat

 Linked

 47
 How does Spark aggregate function - aggregateByKey work?

 Related

 863
 Most efficient method to groupby on an array of objects
 380
 Apply multiple functions to multiple groupby columns
 749
 Get statistics for each group (such as count, mean, etc) using pandas GroupBy?
 180
 How can I change column types in Spark SQL's DataFrame?
 174
 How to sort by column in descending order in Spark SQL?
 335
 Difference between DataFrame, Dataset, and RDD in Spark
 397
 Spark - repartition() vs coalesce()
 4
 Spark - Nested RDD Operation
 3
 Spark DataFrame groupBy
 1
 Spark SQL Join: how it actually works

 Hot Network Questions

 * Why is the solstice the first day of summer, and not the central day?
 * How does putting an LED in parallel with an 8-bit computer load affect the voltage/current of a circuit?
 * Do conference organizers make a good profit?
 * Why cartilaginous fishes do not excrete ammonia?
 * Do astronauts use plumber's bifocals?
 * Did JK Rowling ever describe Harry Potter goblins as "hook-nosed"?
 * How does one Calculate Dual Variable Probability?
 * How to run `usepackage{times}` while keeping texttt font the same
 * I forgot to include $10 in interest in my tax return. My tax liability did not change. Should I file an amended return?
 * When did Kievan Rus' rulers become culturally Slavic?
 * Black-and-white SF movie with giant octopus- or starfish-like creature attacking a city
 * Is the set of real matrices with at least one real logarithm closed under multiplication?
 * How does Freedom of Movement interact with magical effects which increase movement costs?
 * Is there more dark matter than we previously thought?
 * Why is velocity Verlet better than Verlet for gravity if it has a worse order of magnitude for the error term
 * Distance to the average of the next two prime numbers
 * The conjunction "autem" expressing indignation or surprise in questions and exclamations
 * Find the perimeter (seemingly unsolvable problem)
 * Jhoira, Ageless Innovator and Prototype Artifacts
 * The stock brakes on many e-bikes may be inadequate. What changes would make them adequate, and at what cost?
 * Put a number on both sides
 * At what point does combining integrated circuits yield something patentable?
 * Why does sys.fn_xe_file_target_read_file require an explicit cast on datetime2 column?
 * Reverse the polyglot, change the language
 more hot questions
 Question feed

 Subscribe to RSS

 Question feed

 To subscribe to this RSS feed, copy and paste this URL into your RSS reader.

 Stack Overflow

 * Questions
 * Help

 Products

 * Teams
 * Advertising
 * Collectives
 * Talent

 Company

 * About
 * Press
 * Work Here
 * Legal
 * Privacy Policy
 * Terms of Service
 * Contact Us
 * Cookie Settings
 * Cookie Policy

 Stack Exchange Network

 * Technology
 * Culture & recreation
 * Life & arts
 * Science
 * Professional
 * Business
 * API
 * Data
 * Blog
 * Facebook
 * Twitter
 * LinkedIn
 * Instagram

 Site design / logo © 2023 Stack Exchange Inc; user contributions licensed under CC BY-SA. rev 2023.2.7.43225

 Your privacy

 By clicking “Accept all cookies”, you agree Stack Exchange can store cookies on your device and disclose information in accordance with our Cookie Policy.

 Accept all cookies Customize settings
We are facing poor performance using Spark.
I have 2 specific questions:
When debugging we noticed that a few of the groupby operations done on Rdd are taking more time 
Also a few of the stages are appearing twice, some finishing very quickly, some taking more time 
Here is a screenshot of .
Currently running locally, having shuffle partitions set to 2 and number of partitions set to 5, data is around 1,00,000 records.
Speaking of groupby operation, we are grouping a dataframe (which is a result of several joins) based on two columns, and then applying a function to get some result.
  val groupedRows = rows.rdd.groupBy(row =&gt; (
 row.getAs[Long](Column1),
 row.getAs[Int](Column2)
 ))
 val rdd = groupedRows.values.map(Criteria)
 
Where Criteria is some function acted on the grouped resultant rows. Can we optimize this group by in any way?
Here is a screenshot of the .
Improve groupby operation in Spark 1.5.2
https://stackoverflow.com/questions/44734088/improve-groupby-operation-in-spark-1-5-2
https://stackoverflow.com/questions/52274221/why-is-groupby-a-lot-faster-than-distinct-in-pyspark
Why is groupBy() a lot faster than distinct() in pyspark?
2018-09-11
Pri Pri 1
<div><div>
<p>I've recently focused on the difference between the <code>GROUP BY</code> and <code>DISTINCT</code> operations in Apache Spark SQL. It happens that...both can sometimes be the same!</p>
<p>To see this, run the following code and check the execution plans:</p>
<pre><code>(0 to 10).map(id =&gt; (s"id#${id}", s"login${id % 25}"))
 .toDF("id", "login").createTempView("users")
sparkSession.sql("SELECT login FROM users GROUP BY login").explain(true)
sparkSession.sql("SELECT DISTINCT(login) FROM users").explain(true)
</code></pre>
<p>Surprise, surprise! The plans should look like that:</p>
<pre><code>== Physical Plan ==
*(2) HashAggregate(keys=[login#8], functions=[], output=[login#8])
+- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#33]
 +- *(1) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- *(1) LocalTableScan [login#8]
</code></pre>
<p>Why? Because of the <a href="https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L1775">ReplaceDistinctWithAggregate</a> rule that you should see in action in the logs:</p>
<pre><code>=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ReplaceDistinctWithAggregate ===
!Distinct Aggregate [login#8], [login#8]
 +- LocalRelation [login#8] +- LocalRelation [login#8]
 (org.apache.spark.sql.catalyst.rules.PlanChangeLogger:65)
</code></pre>
<p>===========================
Update:</p>
<p>For more complex queries (e.g. with aggregates), it can be a difference.</p>
<pre><code>sparkSession.sql("SELECT COUNT(login) FROM users GROUP BY login").explain(true)
sparkSession.sql("SELECT COUNT(DISTINCT(login)) FROM users").explain(true)
</code></pre>
<p>The <code>GROUP BY</code> version generates a plan with only one shuffle:</p>
<pre><code>== Physical Plan ==
*(2) HashAggregate(keys=[login#8], functions=[count(login#8)], output=[count(login)#12L])
+- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#16]
 +- *(1) HashAggregate(keys=[login#8], functions=[partial_count(login#8)], output=[login#8, count#15L])
 +- *(1) LocalTableScan [login#8]
</code></pre>
<p>Whereas the version with <code>DISTINCT</code> generates 2 shuffles. The first is there to deduplicate the logins and the second to count them:</p>
<pre><code>== Physical Plan ==
*(3) HashAggregate(keys=[], functions=[count(distinct login#8)], output=[count(DISTINCT login)#17L])
+- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#48]
 +- *(2) HashAggregate(keys=[], functions=[partial_count(distinct login#8)], output=[count#21L])
 +- *(2) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#43]
 +- *(1) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- *(1) LocalTableScan [login#8]
</code></pre>
<p>However, semantically these queries are not the same because the first generates the login groups whereas the second also counts them. And it explains the extra shuffle step.</p>
<p>It could be easier to answer the question with the code before/after the change. @pri, do you have it so that we can analyze the plans executed by PySpark?</p>
 </div>||||I|||| Stack Overflow
 1. About
 2. Products
 3. For Teams
 1. Stack Overflow Public questions & answers
 2. Stack Overflow for Teams Where developers & technologists share private knowledge with coworkers
 3. Talent Build your employer brand
 4. Advertising Reach developers & technologists worldwide
 5. About the company
 Loading…

 1. current community

 + Stack Overflow
 help chat
 + Meta Stack Overflow

 your communities

 Sign up or log in to customize your list.

 more stack exchange communities

 company blog
 2. 
 3. Log in
 4. Sign up
 1. Home
 2. 
 1. Public
 2. Questions
 3. Tags
 4. Users
 5. Companies
 6. Collectives
 7. Explore Collectives
 3. 
 1. Teams
 Stack Overflow for Teams – Start collaborating and sharing organizational knowledge. Create a free Team Why Teams?
 2. Teams
 3. Create free Team

 Collectives™ on Stack Overflow

 Find centralized, trusted content and collaborate around the technologies you use most.

 Learn more about Collectives

 Teams

 Q&A for work

 Connect and share knowledge within a single location that is structured and easy to search.

 Learn more about Teams

 Why is groupBy() a lot faster than distinct() in pyspark?

 Ask Question
 Asked 4 years, 4 months ago
 Modified 1 year, 6 months ago
 Viewed 9k times
 10

 I have seen a lot of performance improvement in my pyspark code when I replaced distinct() on a spark data frame with groupBy(). But I failed to understand the reason behind it. The whole intention was to remove the row level duplicates from the dataframe.

 I tried Googling the implementation of groupBy() and distinct() in pyspark, but was unable to find it.

 Can somebody explain or point me in the right direction for the explanation?

 * pyspark
 Share
 Follow
 asked Sep 11, 2018 at 10:49
 pri pri
 1,501 1 1 gold badge 18 18 silver badges 26 26 bronze badges
 Add a comment |

 2 Answers

 Sorted by: Reset to default
 Highest score (default) Trending (recent votes count more) Date modified (newest first) Date created (oldest first)
 4

 I've recently focused on the difference between the GROUP BY and DISTINCT operations in Apache Spark SQL. It happens that...both can sometimes be the same!

 To see this, run the following code and check the execution plans:

 (0 to 10).map(id => (s"id#${id}", s"login${id % 25}"))
 .toDF("id", "login").createTempView("users")
 
 sparkSession.sql("SELECT login FROM users GROUP BY login").explain(true)
 sparkSession.sql("SELECT DISTINCT(login) FROM users").explain(true)
 

 Surprise, surprise! The plans should look like that:

 == Physical Plan ==
 *(2) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#33]
 +- *(1) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- *(1) LocalTableScan [login#8]
 

 Why? Because of the ReplaceDistinctWithAggregate rule that you should see in action in the logs:

 === Applying Rule org.apache.spark.sql.catalyst.optimizer.ReplaceDistinctWithAggregate ===
 !Distinct Aggregate [login#8], [login#8]
 +- LocalRelation [login#8] +- LocalRelation [login#8]
 (org.apache.spark.sql.catalyst.rules.PlanChangeLogger:65)
 

 =========================== Update:

 For more complex queries (e.g. with aggregates), it can be a difference.

 sparkSession.sql("SELECT COUNT(login) FROM users GROUP BY login").explain(true)
 sparkSession.sql("SELECT COUNT(DISTINCT(login)) FROM users").explain(true)
 

 The GROUP BY version generates a plan with only one shuffle:

 == Physical Plan ==
 *(2) HashAggregate(keys=[login#8], functions=[count(login#8)], output=[count(login)#12L])
 +- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#16]
 +- *(1) HashAggregate(keys=[login#8], functions=[partial_count(login#8)], output=[login#8, count#15L])
 +- *(1) LocalTableScan [login#8]
 

 Whereas the version with DISTINCT generates 2 shuffles. The first is there to deduplicate the logins and the second to count them:

 == Physical Plan ==
 *(3) HashAggregate(keys=[], functions=[count(distinct login#8)], output=[count(DISTINCT login)#17L])
 +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#48]
 +- *(2) HashAggregate(keys=[], functions=[partial_count(distinct login#8)], output=[count#21L])
 +- *(2) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#43]
 +- *(1) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- *(1) LocalTableScan [login#8]
 

 However, semantically these queries are not the same because the first generates the login groups whereas the second also counts them. And it explains the extra shuffle step.

 It could be easier to answer the question with the code before/after the change. @pri, do you have it so that we can analyze the plans executed by PySpark?

 Share
 Follow
 edited Jul 16, 2021 at 8:14
 answered Jul 13, 2021 at 16:42
 Bartosz Konieczny Bartosz Konieczny
 1,943 11 11 silver badges 25 25 bronze badges
 2
 * so how to explain the observation then?
 – thebluephantom
 Jul 13, 2021 at 20:33
 * 1
 Analyzing Spark UI, maybe there is something else than only distinct vs group by making the difference? Maybe it applies to specific datasets or queries? Maybe it's just an "opinion" coming from Hive ( stackoverflow.com/a/31876286/9726075 )? No idea. I'll share if I find something new.
 – Bartosz Konieczny
 Jul 14, 2021 at 5:03
 Add a comment |
 -1

 distinct() implementation check every columns and if two or more lines totally same keep the first line. I think this is the main reason, why distinct so slower.

 Check this topic too.

 Share
 Follow
 answered Sep 11, 2018 at 11:19
 Aron Asztalos Aron Asztalos
 764 7 7 silver badges 7 7 bronze badges
 4
 * 1
 Thanks for your answer, but I didn't find any reason which makes groupBy so faster as compared to distinct().
 – pri
 Sep 11, 2018 at 11:25
 * if you have 100+ columns, the spark job should compare all columns. groupBy works with one (usually) column, distinct() check works with all columns. Just imagine that, you have check all columns, how difficult task is.
 – Aron Asztalos
 Sep 11, 2018 at 12:32
 * 1
 If I want to remove row level duplicates, then I'll be including all the columns in the groupBy as well, so that shouldn't matter.
 – pri
 Sep 11, 2018 at 12:42
 * 1
 If you use groupby() executors will makes the grouping, after send the groups to the master which only do the sum, count, etc by group however distinct() check every columns in executors() and try to drop the duplicates after the executors sends the distinct dataframes to the master, and the master check again the distinct values with the all columns. I think this is the main reason.
 – Aron Asztalos
 Sep 11, 2018 at 13:22
 Add a comment |

 Your Answer

 Thanks for contributing an answer to Stack Overflow!

 * Please be sure to answer the question. Provide details and share your research!

 But avoid …

 * Asking for help, clarification, or responding to other answers.
 * Making statements based on opinion; back them up with references or personal experience.

 To learn more, see our tips on writing great answers.

 Sign up or log in

 Sign up using Google
 Sign up using Facebook
 Sign up using Email and Password
 Submit

 Post as a guest

 Name
 Email

 Required, but never shown

 Post as a guest

 Name
 Email

 Required, but never shown

 Post Your Answer Discard

 By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

 Not the answer you're looking for? Browse other questions tagged
 * pyspark
 or ask your own question.

 * The Overflow Blog
 * Why developer experience is the key to better software, straight from the...
 * The frontier of computing: Q&A with IBM quantum computing research
 * Featured on Meta
 * Accessibility Update: Colors
 * 2022: a year in moderation
 * Should we burninate the [balance] tag?
 * Collectives: The next iteration
 * Temporary policy: ChatGPT is banned

 Linked

 22
 How to get distinct rows in dataframe using pyspark?
 28
 distinct vs group by which is better

 Related

 181
 Show distinct column values in pyspark dataframe
 11
 Performance decrease for huge amount of columns. Pyspark
 2
 Groupby and UDF/UDAF in PySpark while maintaining DataFrame structure
 1
 Groupby / Cube and sum values from distinct rows
 0
 Optimizing conversion between PySpark and pandas DataFrames
 1
 Pyspark Distinct count of User Retention using Pivot

 Hot Network Questions

 * I need a seemingly large battery for project containing several sensors, what should I do?
 * Does any non-Catholic denomination maintain Altar Lamp, following Lev 6:12-13?
 * Help with Graph Consisting of Lines and Nodes
 * Is Metapost still relevant in the age of Lua etc.?
 * He acts as if he owns (vs owned) the place?
 * At what point does combining integrated circuits yield something patentable?
 * Can a helicopter's main rotor mince a human?
 * how to modify a wooden fence to keep nubian goats from extending their heads over and eating dwarf tree on other side
 * How would you say "mindless fun" in German
 * How to adapt a spoke salvaged from a 26" wheel to a 24" one?
 * How to download only certain columns of a binary file using wget?
 * Are there any Oxford Colleges which will accommodate a family including children
 * Better midi playback on mac
 * Ada and its candidate languages
 * Is there a cutoff for Newtonian gravity?
 * Is 'comme platine' an idiom
 * Can an aircraft use ram air to generate power while flying?
 * Addition and multiplication are commutative but exponentiation and tetration are not. Do we know why?
 * What is the use of alumium profile with LED strip lighting?
 * I forgot to include $10 in interest in my tax return. My tax liability did not change. Should I file an amended return?
 * What do you do when you're stuck?
 * Intel 80188 & 8087 clock frequency differences
 * Can "né" or "née" be used to refer to a religious's secular name?
 * Why not use a stablecoin as the base currency of a blockchain?
 more hot questions
 Question feed

 Subscribe to RSS

 Question feed

 To subscribe to this RSS feed, copy and paste this URL into your RSS reader.

 Stack Overflow

 * Questions
 * Help

 Products

 * Teams
 * Advertising
 * Collectives
 * Talent

 Company

 * About
 * Press
 * Work Here
 * Legal
 * Privacy Policy
 * Terms of Service
 * Contact Us
 * Cookie Settings
 * Cookie Policy

 Stack Exchange Network

 * Technology
 * Culture & recreation
 * Life & arts
 * Science
 * Professional
 * Business
 * API
 * Data
 * Blog
 * Facebook
 * Twitter
 * LinkedIn
 * Instagram

 Site design / logo © 2023 Stack Exchange Inc; user contributions licensed under CC BY-SA. rev 2023.2.6.43222

 Your privacy

 By clicking “Accept all cookies”, you agree Stack Exchange can store cookies on your device and disclose information in accordance with our Cookie Policy.

 Accept all cookies Customize settings
I've recently focused on the difference between the GROUP BY and DISTINCT operations in Apache Spark SQL. It happens that...both can sometimes be the same!
To see this, run the following code and check the execution plans:
 (0 to 10).map(id =&gt; (s"id#${id}", s"login${id % 25}"))
 .toDF("id", "login").createTempView("users")
sparkSession.sql("SELECT login FROM users GROUP BY login").explain(true)
sparkSession.sql("SELECT DISTINCT(login) FROM users").explain(true)
 
Surprise, surprise! The plans should look like that:
 == Physical Plan ==
*(2) HashAggregate(keys=[login#8], functions=[], output=[login#8])
+- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#33]
 +- *(1) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- *(1) LocalTableScan [login#8]
 
Why? Because of the ReplaceDistinctWithAggregate rule that you should see in action in the logs:
 === Applying Rule org.apache.spark.sql.catalyst.optimizer.ReplaceDistinctWithAggregate ===
!Distinct Aggregate [login#8], [login#8]
 +- LocalRelation [login#8] +- LocalRelation [login#8]
 (org.apache.spark.sql.catalyst.rules.PlanChangeLogger:65)
 
===========================
Update:
For more complex queries (e.g. with aggregates), it can be a difference.
 sparkSession.sql("SELECT COUNT(login) FROM users GROUP BY login").explain(true)
sparkSession.sql("SELECT COUNT(DISTINCT(login)) FROM users").explain(true)
 
The GROUP BY version generates a plan with only one shuffle:
 == Physical Plan ==
*(2) HashAggregate(keys=[login#8], functions=[count(login#8)], output=[count(login)#12L])
+- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#16]
 +- *(1) HashAggregate(keys=[login#8], functions=[partial_count(login#8)], output=[login#8, count#15L])
 +- *(1) LocalTableScan [login#8]
 
Whereas the version with DISTINCT generates 2 shuffles. The first is there to deduplicate the logins and the second to count them:
 == Physical Plan ==
*(3) HashAggregate(keys=[], functions=[count(distinct login#8)], output=[count(DISTINCT login)#17L])
+- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#48]
 +- *(2) HashAggregate(keys=[], functions=[partial_count(distinct login#8)], output=[count#21L])
 +- *(2) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- Exchange hashpartitioning(login#8, 200), ENSURE_REQUIREMENTS, [id=#43]
 +- *(1) HashAggregate(keys=[login#8], functions=[], output=[login#8])
 +- *(1) LocalTableScan [login#8]
 
However, semantically these queries are not the same because the first generates the login groups whereas the second also counts them. And it explains the extra shuffle step.
It could be easier to answer the question with the code before/after the change. @pri, do you have it so that we can analyze the plans executed by PySpark?
Why is groupBy() a lot faster than distinct() in pyspark?
https://stackoverflow.com/questions/52274221/why-is-groupby-a-lot-faster-than-distinct-in-pyspark
https://stackoverflow.com/questions/7943957/huge-performance-difference-when-using-group-by-vs-distinct
Huge performance difference when using GROUP BY vs DISTINCT
2021-07-07
Martin Dimitrov Martin Dimitrov 4
<div><div>
 
<p>I am performing some tests on a <code>HSQLDB</code> server with a table containing 500 000 entries. The table has no indices. There are 5000 distinct business keys. I need a list of them.</p>
<p>Naturally I started with a <code>DISTINCT</code> query:</p>
<pre><code>SELECT DISTINCT business_key
FROM memory
WHERE concept &lt;&gt; 'case' OR 
 attrib &lt;&gt; 'status' OR 
 value &lt;&gt; 'closed';
</code></pre>
<p>It takes around 90 seconds!!!</p>
<p>Then I tried using <code>GROUP BY</code>:</p>
<pre><code>SELECT business_key
FROM memory
WHERE concept &lt;&gt; 'case' OR 
 attrib &lt;&gt; 'status' OR
 value &lt;&gt; 'closed';
GROUP BY business_key
</code></pre>
<p>And it takes 1 second!!!</p>
<p>Trying to figure out the difference I ran <code>EXLAIN PLAN FOR</code> but it seems to give the same information for both queries.</p>
<p><code>EXLAIN PLAN FOR DISTINCT ...</code></p>
<pre><code>isAggregated=[false]
columns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY
]
[range variable 1
 join type=INNER
 table=MEMORY
 alias=M
 access=FULL SCAN
 condition = [ index=SYS_IDX_SYS_PK_10057_10058
 other condition=[
 OR arg_left=[
 OR arg_left=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.CONCEPT] arg_right=[
 VALUE = case, TYPE = CHARACTER]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.ATTRIB] arg_right=[
 VALUE = status, TYPE = CHARACTER]]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.VALUE] arg_right=[
 VALUE = closed, TYPE = CHARACTER]]]
 ]
]]
PARAMETERS=[]
SUBQUERIES[]
Object References
PUBLIC.MEMORY
PUBLIC.MEMORY.CONCEPT
PUBLIC.MEMORY.ATTRIB
PUBLIC.MEMORY.VALUE
PUBLIC.MEMORY.BUSINESS_KEY
Read Locks
PUBLIC.MEMORY
WriteLocks
</code></pre>
<p><code>EXLAIN PLAN FOR SELECT ... GROUP BY ...</code></p>
<pre><code>isDistinctSelect=[false]
isGrouped=[true]
isAggregated=[false]
columns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY
]
[range variable 1
 join type=INNER
 table=MEMORY
 alias=M
 access=FULL SCAN
 condition = [ index=SYS_IDX_SYS_PK_10057_10058
 other condition=[
 OR arg_left=[
 OR arg_left=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.CONCEPT] arg_right=[
 VALUE = case, TYPE = CHARACTER]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.ATTRIB] arg_right=[
 VALUE = status, TYPE = CHARACTER]]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.VALUE] arg_right=[
 VALUE = closed, TYPE = CHARACTER]]]
 ]
]]
groupColumns=[
COLUMN: PUBLIC.MEMORY.BUSINESS_KEY]
PARAMETERS=[]
SUBQUERIES[]
Object References
PUBLIC.MEMORY
PUBLIC.MEMORY.CONCEPT
PUBLIC.MEMORY.ATTRIB
PUBLIC.MEMORY.VALUE
PUBLIC.MEMORY.BUSINESS_KEY
Read Locks
PUBLIC.MEMORY
WriteLocks
</code></pre>
<hr />
<h2>EDIT</h2>
<p>I did additional tests. With 500 000 records in <code>HSQLDB</code> with all distinct business keys, the performance of <code>DISTINCT</code> is now better - 3 seconds, vs <code>GROUP BY</code> which took around 9 seconds.</p>
<p>In <code>MySQL</code> both queries preform the same:</p>
<p>MySQL: 500 000 rows - 5 000 distinct business keys:
Both queries: 0.5 second
MySQL: 500 000 rows - all distinct business keys:
<code>SELECT DISTINCT ...</code> - 11 seconds
<code>SELECT ... GROUP BY business_key</code> - 13 seconds</p>
<p>So the problem is only related to <code>HSQLDB</code>.</p>
<p>I will be very grateful if someone can explain why there is such a drastic difference.</p>
 </div>||||I|||| Stack Overflow
 1. About
 2. Products
 3. For Teams
 1. Stack Overflow Public questions & answers
 2. Stack Overflow for Teams Where developers & technologists share private knowledge with coworkers
 3. Talent Build your employer brand
 4. Advertising Reach developers & technologists worldwide
 5. About the company
 Loading…

 1. current community

 + Stack Overflow
 help chat
 + Meta Stack Overflow

 your communities

 Sign up or log in to customize your list.

 more stack exchange communities

 company blog
 2. 
 3. Log in
 4. Sign up
 1. Home
 2. 
 1. Public
 2. Questions
 3. Tags
 4. Users
 5. Companies
 6. Collectives
 7. Explore Collectives
 3. 
 1. Teams
 Stack Overflow for Teams – Start collaborating and sharing organizational knowledge. Create a free Team Why Teams?
 2. Teams
 3. Create free Team

 Collectives™ on Stack Overflow

 Find centralized, trusted content and collaborate around the technologies you use most.

 Learn more about Collectives

 Teams

 Q&A for work

 Connect and share knowledge within a single location that is structured and easy to search.

 Learn more about Teams

 Huge performance difference when using GROUP BY vs DISTINCT

 Ask Question
 Asked 11 years, 3 months ago
 Modified 1 year, 7 months ago
 Viewed 145k times
 90

 I am performing some tests on a HSQLDB server with a table containing 500 000 entries. The table has no indices. There are 5000 distinct business keys. I need a list of them.

 Naturally I started with a DISTINCT query:

 SELECT DISTINCT business_key
 FROM memory
 WHERE concept <> 'case' OR 
 attrib <> 'status' OR 
 value <> 'closed';
 

 It takes around 90 seconds!!!

 Then I tried using GROUP BY:

 SELECT business_key
 FROM memory
 WHERE concept <> 'case' OR 
 attrib <> 'status' OR
 value <> 'closed';
 GROUP BY business_key
 

 And it takes 1 second!!!

 Trying to figure out the difference I ran EXLAIN PLAN FOR but it seems to give the same information for both queries.

 EXLAIN PLAN FOR DISTINCT ...

 isAggregated=[false]
 columns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY
 ]
 [range variable 1
 join type=INNER
 table=MEMORY
 alias=M
 access=FULL SCAN
 condition = [ index=SYS_IDX_SYS_PK_10057_10058
 other condition=[
 OR arg_left=[
 OR arg_left=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.CONCEPT] arg_right=[
 VALUE = case, TYPE = CHARACTER]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.ATTRIB] arg_right=[
 VALUE = status, TYPE = CHARACTER]]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.VALUE] arg_right=[
 VALUE = closed, TYPE = CHARACTER]]]
 ]
 ]]
 PARAMETERS=[]
 SUBQUERIES[]
 Object References
 PUBLIC.MEMORY
 PUBLIC.MEMORY.CONCEPT
 PUBLIC.MEMORY.ATTRIB
 PUBLIC.MEMORY.VALUE
 PUBLIC.MEMORY.BUSINESS_KEY
 Read Locks
 PUBLIC.MEMORY
 WriteLocks
 

 EXLAIN PLAN FOR SELECT ... GROUP BY ...

 isDistinctSelect=[false]
 isGrouped=[true]
 isAggregated=[false]
 columns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY
 ]
 [range variable 1
 join type=INNER
 table=MEMORY
 alias=M
 access=FULL SCAN
 condition = [ index=SYS_IDX_SYS_PK_10057_10058
 other condition=[
 OR arg_left=[
 OR arg_left=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.CONCEPT] arg_right=[
 VALUE = case, TYPE = CHARACTER]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.ATTRIB] arg_right=[
 VALUE = status, TYPE = CHARACTER]]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.VALUE] arg_right=[
 VALUE = closed, TYPE = CHARACTER]]]
 ]
 ]]
 groupColumns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY]
 PARAMETERS=[]
 SUBQUERIES[]
 Object References
 PUBLIC.MEMORY
 PUBLIC.MEMORY.CONCEPT
 PUBLIC.MEMORY.ATTRIB
 PUBLIC.MEMORY.VALUE
 PUBLIC.MEMORY.BUSINESS_KEY
 Read Locks
 PUBLIC.MEMORY
 WriteLocks
 

 EDIT

 I did additional tests. With 500 000 records in HSQLDB with all distinct business keys, the performance of DISTINCT is now better - 3 seconds, vs GROUP BY which took around 9 seconds.

 In MySQL both queries preform the same:

 MySQL: 500 000 rows - 5 000 distinct business keys: Both queries: 0.5 second MySQL: 500 000 rows - all distinct business keys: SELECT DISTINCT ... - 11 seconds SELECT ... GROUP BY business_key - 13 seconds

 So the problem is only related to HSQLDB.

 I will be very grateful if someone can explain why there is such a drastic difference.

 * sql
 * performance
 * group-by
 * distinct
 * hsqldb
 Share
 Improve this question
 Follow
 edited Jul 7, 2021 at 0:06
 informatik01
 15.8k 10 10 gold badges 74 74 silver badges 103 103 bronze badges
 asked Oct 30, 2011 at 8:29
 Martin Dimitrov Martin Dimitrov
 4,766 4 4 gold badges 46 46 silver badges 58 58 bronze badges
 6
 * 3
 please show the result of EXPLAIN PLAN AND try running the DISTINCT query AFTER you run the GROUP BY to see whether perhaps some caching is skewing the timing...
 – Yahia
 Oct 30, 2011 at 8:31
 * Given you get the same plan for each query, it sounds like either the table data or the result has been cached.
 – a'r
 Oct 30, 2011 at 8:32
 * I ran them so many times that believe caching is not an issue. I am posting the EXLAIN PLAN FOR output.
 – Martin Dimitrov
 Oct 30, 2011 at 8:36
 * I have an idea, but I am really not sure - please try SELECT DISTINCT business_key FROM (SELECT business_key FROM memory WHERE concept <> 'case' or attrib <> 'status' or value <> 'closed') - this ought to show the same performance you see with the GROUP BY IF my idea is right.
 – Yahia
 Oct 30, 2011 at 9:08
 * @Yahia: still very slow - 94 seconds. I will run the same queries in MySQL to see what will show
 – Martin Dimitrov
 Oct 30, 2011 at 9:37
 | Show 1 more comment

 1 Answer

 Sorted by: Reset to default
 Highest score (default) Trending (recent votes count more) Date modified (newest first) Date created (oldest first)
 99

 The two queries express the same question. Apparently the query optimizer chooses two different execution plans. My guess would be that the distinct approach is executed like:

 * Copy all business_key values to a temporary table
 * Sort the temporary table
 * Scan the temporary table, returning each item that is different from the one before it

 The group by could be executed like:

 * Scan the full table, storing each value of business key in a hashtable
 * Return the keys of the hashtable

 The first method optimizes for memory usage: it would still perform reasonably well when part of the temporary table has to be swapped out. The second method optimizes for speed, but potentially requires a large amount of memory if there are a lot of different keys.

 Since you either have enough memory or few different keys, the second method outperforms the first. It's not unusual to see performance differences of 10x or even 100x between two execution plans.

 Share
 Improve this answer
 Follow
 edited Sep 21, 2017 at 19:02
 Andrew H
 5 4 4 bronze badges
 answered Oct 30, 2011 at 10:44
 Andomar Andomar
 230k 47 47 gold badges 371 371 silver badges 396 396 bronze badges
 8
 * Thanks for the reply. Are your guesses evident from the EXPLAIN output? Both look the same to me.
 – Martin Dimitrov
 Oct 30, 2011 at 10:54
 * As far as I can see, the plan doesn't specify how it will execute the join. I'm not even sure why it would execute a join. It probably takes an HSQLDB specialist to read the explain output.
 – Andomar
 Oct 30, 2011 at 11:03
 * As the answer indicates, the second method uses more memory and may hit garbage collection (GC) too often. If you increase the JVM memory allocation, there shouldn't be a huge difference between the two query times.
 – fredt
 Oct 30, 2011 at 21:23
 * I did additional test by entering all distinct keys in the table (see above). Do you thing the result proves your point? Thanks a lot.
 – Martin Dimitrov
 Oct 31, 2011 at 7:07
 * 3
 Can a SME- expert please explain this in more details with examples... I've had this issue a many times but don't seem to get around it... I know the fix but I want to know how and WHY
 – singhswat
 Dec 3, 2015 at 13:08
 | Show 3 more comments

 Your Answer

 Thanks for contributing an answer to Stack Overflow!

 * Please be sure to answer the question. Provide details and share your research!

 But avoid …

 * Asking for help, clarification, or responding to other answers.
 * Making statements based on opinion; back them up with references or personal experience.

 To learn more, see our tips on writing great answers.

 Sign up or log in

 Sign up using Google
 Sign up using Facebook
 Sign up using Email and Password
 Submit

 Post as a guest

 Name
 Email

 Required, but never shown

 Post as a guest

 Name
 Email

 Required, but never shown

 Post Your Answer Discard

 By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

 Not the answer you're looking for? Browse other questions tagged
 * sql
 * performance
 * group-by
 * distinct
 * hsqldb
 or ask your own question.

 * The Overflow Blog
 * The nature of simulating nature: Q&A with IBM quantum computing research
 * The AI that writes music from text (Ep. 535)
 * Featured on Meta
 * Accessibility Update: Colors
 * Should we burninate the [balance] tag?
 * Collectives: The next iteration
 * Temporary policy: ChatGPT is banned

 Linked

 119
 How to display unique records from a has_many through relationship?
 27
 SQL Performance: SELECT DISTINCT versus GROUP BY
 2
 Faster way compare changes between SQL sub-queries?
 2
 (Oracle) SQL: Column must contain a list of values
 2
 I want to select distinct values from the table
 0
 Distinct query causing the query to be slow
 2
 MYSQL JOIN and GROUP / DISTINCT
 1
 SQLite query using CursorLoader
 1
 SQL comparison between 3 columns
 0
 select distinct values of a replaced selected result
 See more linked questions

 Related

 524
 Is there a performance difference between i++ and ++i in C?
 438
 Is there any difference between GROUP BY and DISTINCT
 315
 What's faster, SELECT DISTINCT or GROUP BY in MySQL?
 499
 Using LIMIT within GROUP BY to get N results per group?
 1448
 Using group by on multiple columns
 1
 Optimizing DISTINCT SQL query with OR conditions
 0
 Getting different results from group by and distinct
 749
 Get statistics for each group (such as count, mean, etc) using pandas GroupBy?
 0
 Linq Group by / Distinct with Join table
 2
 Counting distinct values within group

 Hot Network Questions

 * Will any Oxford College accommodate a family including children?
 * When did Kievan Rus' rulers become culturally Slavic?
 * Why cartilaginous fishes do not excrete ammonia?
 * Why does sys.fn_xe_file_target_read_file require an explicit cast on datetime2 column?
 * How to halve DC motor speed without using PWM
 * Do conference organizers make a good profit?
 * Why is mind interacting with matter any more problematic than matter interacting with matter?
 * What exactly is the "clutch" doing on a Shimano M8100 rear derailleur?
 * What do you do when you're stuck at a problem?
 * How to generate 3-subsets that satisfy certain conditions without post-filtering?
 * Can a helicopter's main rotor mince a human?
 * What is the default visibility of a variable in solidity?
 * A journal editor has pretended my referee report actually came from two separate referees
 * Is the set of real matrices with at least one real logarithm closed under multiplication?
 * Black-and-white SF movie with giant octopus- or starfish-like creature attacking a city
 * Searching text about school girl growing wings and trying to hide them at first
 * Which Riemann tensor should I choose?
 * The stock brakes on many e-bikes may be inadequate. What changes would make them adequate, and at what cost?
 * Why arm cortex series cpu don't list their cpu speed in Hz?
 * Is Chimpanzee mitochondrial DNA identical to human?
 * How to play a measure with no chord notation above it?
 * What do you do when you're stuck?
 * How would a photon-based intelligence work in regard to relativity?
 * According to Catholic teaching, is a priest who denies the sin of homosexuality and affirms transgender identity a heretic or a dissenter?
 more hot questions
 Question feed

 Subscribe to RSS

 Question feed

 To subscribe to this RSS feed, copy and paste this URL into your RSS reader.

 Stack Overflow

 * Questions
 * Help

 Products

 * Teams
 * Advertising
 * Collectives
 * Talent

 Company

 * About
 * Press
 * Work Here
 * Legal
 * Privacy Policy
 * Terms of Service
 * Contact Us
 * Cookie Settings
 * Cookie Policy

 Stack Exchange Network

 * Technology
 * Culture & recreation
 * Life & arts
 * Science
 * Professional
 * Business
 * API
 * Data
 * Blog
 * Facebook
 * Twitter
 * LinkedIn
 * Instagram

 Site design / logo © 2023 Stack Exchange Inc; user contributions licensed under CC BY-SA. rev 2023.2.7.43225

 Your privacy

 By clicking “Accept all cookies”, you agree Stack Exchange can store cookies on your device and disclose information in accordance with our Cookie Policy.

 Accept all cookies Customize settings
I am performing some tests on a HSQLDB server with a table containing 500 000 entries. The table has no indices. There are 5000 distinct business keys. I need a list of them.
Naturally I started with a DISTINCT query:
 SELECT DISTINCT business_key
FROM memory
WHERE concept &lt;&gt; 'case' OR 
 attrib &lt;&gt; 'status' OR 
 value &lt;&gt; 'closed';
 
It takes around 90 seconds!!!
Then I tried using GROUP BY:
 SELECT business_key
FROM memory
WHERE concept &lt;&gt; 'case' OR 
 attrib &lt;&gt; 'status' OR
 value &lt;&gt; 'closed';
GROUP BY business_key
 
And it takes 1 second!!!
Trying to figure out the difference I ran EXLAIN PLAN FOR but it seems to give the same information for both queries.
 EXLAIN PLAN FOR DISTINCT ... 
 isAggregated=[false]
columns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY
]
[range variable 1
 join type=INNER
 table=MEMORY
 alias=M
 access=FULL SCAN
 condition = [ index=SYS_IDX_SYS_PK_10057_10058
 other condition=[
 OR arg_left=[
 OR arg_left=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.CONCEPT] arg_right=[
 VALUE = case, TYPE = CHARACTER]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.ATTRIB] arg_right=[
 VALUE = status, TYPE = CHARACTER]]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.VALUE] arg_right=[
 VALUE = closed, TYPE = CHARACTER]]]
 ]
]]
PARAMETERS=[]
SUBQUERIES[]
Object References
PUBLIC.MEMORY
PUBLIC.MEMORY.CONCEPT
PUBLIC.MEMORY.ATTRIB
PUBLIC.MEMORY.VALUE
PUBLIC.MEMORY.BUSINESS_KEY
Read Locks
PUBLIC.MEMORY
WriteLocks
 
 EXLAIN PLAN FOR SELECT ... GROUP BY ... 
 isDistinctSelect=[false]
isGrouped=[true]
isAggregated=[false]
columns=[
 COLUMN: PUBLIC.MEMORY.BUSINESS_KEY
]
[range variable 1
 join type=INNER
 table=MEMORY
 alias=M
 access=FULL SCAN
 condition = [ index=SYS_IDX_SYS_PK_10057_10058
 other condition=[
 OR arg_left=[
 OR arg_left=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.CONCEPT] arg_right=[
 VALUE = case, TYPE = CHARACTER]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.ATTRIB] arg_right=[
 VALUE = status, TYPE = CHARACTER]]] arg_right=[
 NOT_EQUAL arg_left=[
 COLUMN: PUBLIC.MEMORY.VALUE] arg_right=[
 VALUE = closed, TYPE = CHARACTER]]]
 ]
]]
groupColumns=[
COLUMN: PUBLIC.MEMORY.BUSINESS_KEY]
PARAMETERS=[]
SUBQUERIES[]
Object References
PUBLIC.MEMORY
PUBLIC.MEMORY.CONCEPT
PUBLIC.MEMORY.ATTRIB
PUBLIC.MEMORY.VALUE
PUBLIC.MEMORY.BUSINESS_KEY
Read Locks
PUBLIC.MEMORY
WriteLocks
 
EDIT
I did additional tests. With 500 000 records in HSQLDB with all distinct business keys, the performance of DISTINCT is now better - 3 seconds, vs GROUP BY which took around 9 seconds.
In MySQL both queries preform the same:
MySQL: 500 000 rows - 5 000 distinct business keys:
Both queries: 0.5 second
MySQL: 500 000 rows - all distinct business keys:
SELECT DISTINCT ... - 11 seconds
SELECT ... GROUP BY business_key - 13 seconds
So the problem is only related to HSQLDB.
I will be very grateful if someone can explain why there is such a drastic difference.
Huge performance difference when using GROUP BY vs DISTINCT
https://stackoverflow.com/questions/7943957/huge-performance-difference-when-using-group-by-vs-distinct
https://gist.github.com/EricSchles/51796e7b834c936551235257651df8d4
custom_pyspark_groupby.py
2022-03-28
EricSchles
<div><div>
 
 <div>
 
 
 
 
 This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
 <a href="https://github.co/hiddenchars">Learn more about bidirectional Unicode characters</a>
 
 <div> <a href="{{ revealButtonHref }}"> Show hidden characters
</a>
</div>
</div>
 
 
 
 <table>
 <tbody><tr>
 <td></td>
 <td>from pyspark.sql.functions import col</td>
 </tr>
 <tr>
 <td></td>
 <td>
</td>
 </tr>
 <tr>
 <td></td>
 <td>def groupby(df, columns):</td>
 </tr>
 <tr>
 <td></td>
 <td> sdf = df.to_spark()</td>
 </tr>
 <tr>
 <td></td>
 <td> _groups = sdf.select(*columns).distinct().collect()</td>
 </tr>
 <tr>
 <td></td>
 <td> _groups = [group.asDict() for group in _groups]</td>
 </tr>
 <tr>
 <td></td>
 <td> groups = []</td>
 </tr>
 <tr>
 <td></td>
 <td> for group in _groups:</td>
 </tr>
 <tr>
 <td></td>
 <td> tmp = []</td>
 </tr>
 <tr>
 <td></td>
 <td> for column in columns:</td>
 </tr>
 <tr>
 <td></td>
 <td> tmp.append(group[column])</td>
 </tr>
 <tr>
 <td></td>
 <td> groups.append(tmp)</td>
 </tr>
 <tr>
 <td></td>
 <td> </td>
 </tr>
 <tr>
 <td></td>
 <td> for group in groups:</td>
 </tr>
 <tr>
 <td></td>
 <td> mask = (col(columns[0]) == group[0])</td>
 </tr>
 <tr>
 <td></td>
 <td> for index, column in enunerate(columns):</td>
 </tr>
 <tr>
 <td></td>
 <td> if index == 0:</td>
 </tr>
 <tr>
 <td></td>
 <td> continue</td>
 </tr>
 <tr>
 <td></td>
 <td> mask &amp;= (col(column) == group[index])</td>
 </tr>
 <tr>
 <td></td>
 <td> yield (group, sdf.filter(mask).to_pandas_on_spark())</td>
 </tr>
 <tr>
 <td></td>
 <td> </td>
 </tr>
 <tr>
 <td></td>
 <td> </td>
 </tr>
 </tbody></table>
</div>||||I|||| Skip to content
 All gists Back to GitHub Sign in Sign up
 Sign in Sign up
 {{ message }}

 Instantly share code, notes, and snippets.

 EricSchles / custom_pyspark_groupby.py

 Created March 28, 2022 13:38
 * Star 0
 * Fork 0
 Star
 Code Revisions 1
 Embed
 What would you like to do?
 Embed Embed this gist in your website.
 Share Copy sharable link for this gist.
 Clone via HTTPS Clone with Git or checkout with SVN using the repository’s web address.
 Learn more about clone URLs
 Download ZIP
 Raw
 custom_pyspark_groupby.py
 This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
 Show hidden characters
 from pyspark . sql . functions import col 
  
 def groupby ( df , columns ): 
 sdf = df . to_spark () 
 _groups = sdf . select ( * columns ). distinct (). collect ()
 _groups = [ group . asDict () for group in _groups ] 
 groups = [] 
 for group in _groups : 
 tmp = [] 
 for column in columns : 
 tmp . append ( group [ column ]) 
 groups . append ( tmp ) 
  
 for group in groups : 
 mask = ( col ( columns [ 0 ]) == group [ 0 ]) 
 for index , column in enunerate ( columns ): 
 if index == 0 : 
 continue 
 mask &= ( col ( column ) == group [ index ]) 
 yield ( group , sdf . filter ( mask ). to_pandas_on_spark ())
  
  
 
 Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

 Footer

 © 2023 GitHub, Inc.

 Footer navigation

 * Terms
 * Privacy
 * Security
 * Status
 * Docs
 * Contact GitHub
 * Pricing
 * API
 * Training
 * Blog
 * About
 You can’t perform that action at this time.
 You signed in with another tab or window. Reload to refresh your session. You signed out in another tab or window. Reload to refresh your session.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
 Learn more about bidirectional Unicode characters
 
   Show hidden characters
 
 
 
 
  
  
 from pyspark.sql.functions import col
 
 
  
 
 
 
  
 def groupby(df, columns):
 
 
  
  sdf = df.to_spark()
 
 
  
  _groups = sdf.select(*columns).distinct().collect()
 
 
  
  _groups = [group.asDict() for group in _groups]
 
 
  
  groups = []
 
 
  
  for group in _groups:
 
 
  
  tmp = []
 
 
  
  for column in columns:
 
 
  
  tmp.append(group[column])
 
 
  
  groups.append(tmp)
 
 
  
  
 
 
  
  for group in groups:
 
 
  
  mask = (col(columns[0]) == group[0])
 
 
  
  for index, column in enunerate(columns):
 
 
  
  if index == 0:
 
 
  
  continue
 
 
  
  mask &amp;= (col(column) == group[index])
 
 
  
  yield (group, sdf.filter(mask).to_pandas_on_spark())
custom_pyspark_groupby.py
https://gist.github.com/EricSchles/51796e7b834c936551235257651df8d4
https://sparkbyexamples.com/pyspark/pyspark-groupby-explained-with-example
PySpark Groupby Explained with Example - Spark By {Examples}
2020-06-14
Naveen
<div><div>
<p>Similar to SQL <code>GROUP BY</code> clause, PySpark <code>groupBy()</code> function is used to collect the identical data into groups on DataFrame and perform count, sum, avg, min, max functions on the grouped data. In this article, I will explain several <code>groupBy()</code> examples using PySpark (Spark with Python). </p>
<p><strong>Related:</strong> <a href="https://sparkbyexamples.com/spark/using-groupby-on-dataframe/">How to group and aggregate data using Spark and Scala</a></p>
<h2>1. GroupBy() Syntax &amp; Usage</h2>
<p><strong>Syntax:</strong></p>
<pre><code>
# Syntax
DataFrame.groupBy(*cols)
#or 
DataFrame.groupby(*cols)
</code></pre>
<p>When we perform <code>groupBy()</code> on PySpark Dataframe, it returns <code>GroupedData</code> object which contains below aggregate functions.</p>
<p><code>count()</code> – Use <a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-count-explained/">groupBy() count()</a> to return the number of rows for each group.</p>
<p><code>mean()</code> – Returns the mean of values for each group.</p>
<p><code>max()</code> – Returns the maximum of values for each group. </p>
<p><code>min()</code> – Returns the minimum of values for each group.</p>
<p><code>sum()</code> – Returns the total for values for each group. </p>
<p><code>avg()</code> – Returns the average for values for each group. </p>
<p><code>agg()</code> – Using <a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-agg-aggregate-explained/">groupBy() agg()</a> function, we can calculate more than one aggregate at a time.</p>
<p><code>pivot()</code> – This function is used to Pivot the DataFrame which I will not be covered in this article as I already have a dedicated article for<a href="https://sparkbyexamples.com/spark/how-to-pivot-table-and-unpivot-a-spark-dataframe/"> Pivot &amp; Unpivot DataFrame</a>.</p>
<p>Before we start, let’s<a href="https://sparkbyexamples.com/pyspark/different-ways-to-create-dataframe-in-pyspark/"> create the DataFrame</a> from a sequence of the data to work with. This DataFrame contains columns “<code>employee_name</code>”, “<code>department</code>”, “<code>state</code>“, “<code>salary</code>”, “<code>age</code>” and “<code>bonus</code>” columns.</p>
<p>We will use this PySpark DataFrame to run groupBy() on “department” columns and calculate aggregates like minimum, maximum, average, and total salary for each group using min(), max(), and sum() aggregate functions respectively.</p>
<pre><code>
simpleData = [("James","Sales","NY",90000,34,10000),
 ("Michael","Sales","NY",86000,56,20000),
 ("Robert","Sales","CA",81000,30,23000),
 ("Maria","Finance","CA",90000,24,23000),
 ("Raman","Finance","CA",99000,40,24000),
 ("Scott","Finance","NY",83000,36,19000),
 ("Jen","Finance","NY",79000,53,15000),
 ("Jeff","Marketing","CA",80000,25,18000),
 ("Kumar","Marketing","NY",91000,50,21000)
 ]
schema = ["employee_name","department","state","salary","age","bonus"]
df = spark.createDataFrame(data=simpleData, schema = schema)
df.printSchema()
df.show(truncate=False)
</code></pre>
<p>Yields below output.</p><pre><code>
+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
| James| Sales| NY| 90000| 34|10000|
| Michael| Sales| NY| 86000| 56|20000|
| Robert| Sales| CA| 81000| 30|23000|
| Maria| Finance| CA| 90000| 24|23000|
| Raman| Finance| CA| 99000| 40|24000|
| Scott| Finance| NY| 83000| 36|19000|
| Jen| Finance| NY| 79000| 53|15000|
| Jeff| Marketing| CA| 80000| 25|18000|
| Kumar| Marketing| NY| 91000| 50|21000|
+-------------+----------+-----+------+---+-----+
</code></pre>
<h2>2. PySpark groupBy on DataFrame Columns</h2>
<p>Let’s do the <code>groupBy()</code> on <code>department</code> column of DataFrame and then find the sum of salary for each department using <code>sum()</code> function.</p>
<pre><code>
df.groupBy("department").sum("salary").show(truncate=False)
+----------+-----------+
|department|sum(salary)|
+----------+-----------+
|Sales |257000 |
|Finance |351000 |
|Marketing |171000 |
+----------+-----------+
</code></pre>
<p>Similarly, we can calculate the number of employees in each department using.</p>
<pre><code>
df.groupBy("department").count()
</code></pre>
<p>Calculate the minimum salary of each department using <code>min()</code> </p>
<pre><code>
df.groupBy("department").min("salary")
</code></pre>
<p>Calculate the maximin salary of each department using <code>max()</code> </p><pre><code>
df.groupBy("department").max("salary")
</code></pre>
<p>Calculate the average salary of each department using <code>avg()</code> </p>
<pre><code>
df.groupBy("department").avg( "salary")
</code></pre>
<p>Calculate the mean salary of each department using <code>mean()</code> </p>
<pre><code>
df.groupBy("department").mean( "salary") 
</code></pre>
<h2>3. Using Multiple columns </h2>
<p>Similarly, we can also run groupBy and aggregate on two or more DataFrame columns, below example does group by on <code>department</code>,<code>state</code> and does sum() on <code>salary</code> and <code>bonus</code> columns.</p><pre><code>
#GroupBy on multiple columns
df.groupBy("department","state") \
 .sum("salary","bonus") \
 .show(false)
</code></pre>
<p>This yields the below output.</p>
<pre><code>
+----------+-----+-----------+----------+
|department|state|sum(salary)|sum(bonus)|
+----------+-----+-----------+----------+
|Finance |NY |162000 |34000 |
|Marketing |NY |91000 |21000 |
|Sales |CA |81000 |23000 |
|Marketing |CA |80000 |18000 |
|Finance |CA |189000 |47000 |
|Sales |NY |176000 |30000 |
+----------+-----+-----------+----------+
</code></pre>
<p>Similarly, we can run group by and aggregate on two or more columns for other aggregate functions, please refer to the below example.</p>
<h2>4. Running more aggregates at a time</h2>
<p>Using <a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-agg-aggregate-explained/">agg() aggregate function</a> we can calculate many aggregations at a time on a single statement using SQL functions sum(), avg(), min(), max() mean() e.t.c. In order to use these, we should import <code>"from pyspark.sql.functions import sum,avg,max,min,mean,count"</code></p>
<pre><code>
from pyspark.sql.functions import sum,avg,max
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus") \
 ) \
 .show(truncate=False)
</code></pre>
<p>This example does group on <code>department</code> column and calculates <code>sum()</code> and <code>avg()</code> of <code>salary</code> for each department and calculates <code>sum()</code> and <code>max()</code> of bonus for each department.</p>
<pre><code>
+----------+----------+-----------------+---------+---------+
|department|sum_salary|avg_salary |sum_bonus|max_bonus|
+----------+----------+-----------------+---------+---------+
|Sales |257000 |85666.66666666667|53000 |23000 |
|Finance |351000 |87750.0 |81000 |24000 |
|Marketing |171000 |85500.0 |39000 |21000 |
+----------+----------+-----------------+---------+---------+
</code></pre>
<h2>5. Using filter on aggregate data</h2>
<p>Similar to SQL “HAVING” clause, On PySpark DataFrame we can use either <a href="https://sparkbyexamples.com/pyspark/pyspark-dataframe-filter/">where()</a> or <a href="https://sparkbyexamples.com/pyspark/pyspark-dataframe-filter/">filter()</a> function to filter the rows of aggregated data.</p>
<pre><code>
from pyspark.sql.functions import sum,avg,max
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus")) \
 .where(col("sum_bonus") &gt;= 50000) \
 .show(truncate=False)
</code></pre>
<p>This removes the sum of a bonus that has less than 50000 and yields below output.</p>
<pre><code>
+----------+----------+-----------------+---------+---------+
|department|sum_salary|avg_salary |sum_bonus|max_bonus|
+----------+----------+-----------------+---------+---------+
|Sales |257000 |85666.66666666667|53000 |23000 |
|Finance |351000 |87750.0 |81000 |24000 |
+----------+----------+-----------------+---------+---------+
</code></pre>
<h2>6. PySpark groupBy Example Source code</h2>
<pre><code>
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col,sum,avg,max
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
simpleData = [("James","Sales","NY",90000,34,10000),
 ("Michael","Sales","NY",86000,56,20000),
 ("Robert","Sales","CA",81000,30,23000),
 ("Maria","Finance","CA",90000,24,23000),
 ("Raman","Finance","CA",99000,40,24000),
 ("Scott","Finance","NY",83000,36,19000),
 ("Jen","Finance","NY",79000,53,15000),
 ("Jeff","Marketing","CA",80000,25,18000),
 ("Kumar","Marketing","NY",91000,50,21000)
 ]
schema = ["employee_name","department","state","salary","age","bonus"]
df = spark.createDataFrame(data=simpleData, schema = schema)
df.printSchema()
df.show(truncate=False)
df.groupBy("department").sum("salary").show(truncate=False)
df.groupBy("department").count().show(truncate=False)
df.groupBy("department","state") \
 .sum("salary","bonus") \
 .show(truncate=False)
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus") \
 ) \
 .show(truncate=False)
 
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus")) \
 .where(col("sum_bonus") &gt;= 50000) \
 .show(truncate=False)
</code></pre>
<p>This example is also available at <a href="https://github.com/spark-examples/pyspark-examples/blob/master/pyspark-groupby.py">GitHub PySpark Examples</a> project for reference.</p>
<h2>7. Conclusion</h2>
<p>In this tutorial, you have learned how to use <code>groupBy()</code> functions on PySpark DataFrame and also learned how to run these on multiple columns and finally filter data on the aggregated columns.</p>
<p>Thanks for reading. If you like it, please do share the article by following the below social links and any comments or suggestions are welcome in the comments sections!  </p>
<p>Happy Learning !!</p>
<h2>Related Articles</h2>
<ul>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-count-distinct/">PySpark Groupby Count Distinct</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-on-multiple-columns/">PySpark Groupby on Multiple Columns</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-agg-aggregate-explained/">PySpark Groupby Agg (aggregate) – Explained</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-groupby-count-explained/">PySpark GroupBy Count – Explained</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-retrieve-top-n-from-each-group-of-dataframe/">PySpark Select Top N Rows From Each Group</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-find-maximum-row-per-group-in-dataframe/">PySpark Find Maximum Row per Group in DataFrame</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-select-first-row-of-each-group/">PySpark Select First Row of Each Group?</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-column-alias-after-groupby/">PySpark Column alias after groupBy() Example</a></li>
<li><a href="https://sparkbyexamples.com/pyspark/pyspark-dataframe-groupby-and-sort-by-descending-order/">PySpark DataFrame groupBy and Sort by Descending Order</a></li>
</ul>
</div>||||I|||| Skip to content
 * Home
 * About
 * Write For US
 | { One stop for all Spark Examples }
 * 
 * 
 * 
 * 
 * 
 * 
 * Spark
 + Spark RDD
 + Spark DataFrame
 + Spark SQL Functions
 + What’s New in Spark 3.0?
 + Spark Streaming
 + Apache Spark Interview Questions
 * PySpark
 * Pandas
 * R
 + R Programming
 + R Data Frame
 + R dplyr Tutorial
 + R Data Frame
 + R Vector
 + R dplyr Tutorial
 * Snowflake
 * Hive
 * Interview Q
 + Spark Interview Questions
 * More
 + Kafka Apache Kafka Tutorials with Examples
 + NumPy
 + H2O.ai
 + Apache Hadoop
 + Apache HBase
 + Apache Cassandra
 + H2O Sparkling Water
 + Scala Language
 + Python
 * 
 Search this website
 Menu Close
 * Spark
 + Spark RDD
 + Spark DataFrame
 + Spark SQL Functions
 + What’s New in Spark 3.0?
 + Spark Streaming
 + Apache Spark Interview Questions
 * PySpark
 * Pandas
 * R
 + R Programming
 + R Data Frame
 + R dplyr Tutorial
 + R Data Frame
 + R Vector
 + R dplyr Tutorial
 * Snowflake
 * Hive
 * Interview Q
 + Spark Interview Questions
 * More
 + Kafka
 + NumPy
 + H2O.ai
 + Apache Hadoop
 + Apache HBase
 + Apache Cassandra
 + H2O Sparkling Water
 + Scala Language
 + Python
 * 
 * Home
 * About
 * Write For US

 PySpark Groupby Explained with Example

 * Post author: Naveen
 * Post category: PySpark
 * Post last modified: January 10, 2023

 Similar to SQL GROUP BY clause, PySpark groupBy() function is used to collect the identical data into groups on DataFrame and perform count, sum, avg, min, max functions on the grouped data. In this article, I will explain several groupBy() examples using PySpark (Spark with Python).

 Related: How to group and aggregate data using Spark and Scala

 1. GroupBy() Syntax & Usage

 Syntax:

 
 # Syntax
 DataFrame.groupBy(*cols)
 #or 
 DataFrame.groupby(*cols)
 

 When we perform groupBy() on PySpark Dataframe, it returns GroupedData object which contains below aggregate functions.

 count() – Use groupBy() count() to return the number of rows for each group.

 mean() – Returns the mean of values for each group.

 max() – Returns the maximum of values for each group.

 min() – Returns the minimum of values for each group.

 sum() – Returns the total for values for each group.

 avg() – Returns the average for values for each group.

 agg() – Using groupBy() agg() function, we can calculate more than one aggregate at a time.

 pivot() – This function is used to Pivot the DataFrame which I will not be covered in this article as I already have a dedicated article for Pivot & Unpivot DataFrame.

 Before we start, let’s create the DataFrame from a sequence of the data to work with. This DataFrame contains columns “employee_name”, “department”, “state“, “salary”, “age” and “bonus” columns.

 We will use this PySpark DataFrame to run groupBy() on “department” columns and calculate aggregates like minimum, maximum, average, and total salary for each group using min(), max(), and sum() aggregate functions respectively.

 
 simpleData = [("James","Sales","NY",90000,34,10000),
 ("Michael","Sales","NY",86000,56,20000),
 ("Robert","Sales","CA",81000,30,23000),
 ("Maria","Finance","CA",90000,24,23000),
 ("Raman","Finance","CA",99000,40,24000),
 ("Scott","Finance","NY",83000,36,19000),
 ("Jen","Finance","NY",79000,53,15000),
 ("Jeff","Marketing","CA",80000,25,18000),
 ("Kumar","Marketing","NY",91000,50,21000)
 ]
 
 schema = ["employee_name","department","state","salary","age","bonus"]
 df = spark.createDataFrame(data=simpleData, schema = schema)
 df.printSchema()
 df.show(truncate=False)
 

 Yields below output.

 
 +-------------+----------+-----+------+---+-----+
 |employee_name|department|state|salary|age|bonus|
 +-------------+----------+-----+------+---+-----+
 | James| Sales| NY| 90000| 34|10000|
 | Michael| Sales| NY| 86000| 56|20000|
 | Robert| Sales| CA| 81000| 30|23000|
 | Maria| Finance| CA| 90000| 24|23000|
 | Raman| Finance| CA| 99000| 40|24000|
 | Scott| Finance| NY| 83000| 36|19000|
 | Jen| Finance| NY| 79000| 53|15000|
 | Jeff| Marketing| CA| 80000| 25|18000|
 | Kumar| Marketing| NY| 91000| 50|21000|
 +-------------+----------+-----+------+---+-----+
 

 2. PySpark groupBy on DataFrame Columns

 Let’s do the groupBy() on department column of DataFrame and then find the sum of salary for each department using sum() function.

 
 df.groupBy("department").sum("salary").show(truncate=False)
 +----------+-----------+
 |department|sum(salary)|
 +----------+-----------+
 |Sales |257000 |
 |Finance |351000 |
 |Marketing |171000 |
 +----------+-----------+
 

 Similarly, we can calculate the number of employees in each department using.

 
 df.groupBy("department").count()
 

 Calculate the minimum salary of each department using min()

 
 df.groupBy("department").min("salary")
 

 Calculate the maximin salary of each department using max()

 
 df.groupBy("department").max("salary")
 

 Calculate the average salary of each department using avg()

 
 df.groupBy("department").avg( "salary")
 

 Calculate the mean salary of each department using mean()

 
 df.groupBy("department").mean( "salary") 
 

 3. Using Multiple columns

 Similarly, we can also run groupBy and aggregate on two or more DataFrame columns, below example does group by on department,state and does sum() on salary and bonus columns.

 
 #GroupBy on multiple columns
 df.groupBy("department","state") \
 .sum("salary","bonus") \
 .show(false)
 

 This yields the below output.

 
 +----------+-----+-----------+----------+
 |department|state|sum(salary)|sum(bonus)|
 +----------+-----+-----------+----------+
 |Finance |NY |162000 |34000 |
 |Marketing |NY |91000 |21000 |
 |Sales |CA |81000 |23000 |
 |Marketing |CA |80000 |18000 |
 |Finance |CA |189000 |47000 |
 |Sales |NY |176000 |30000 |
 +----------+-----+-----------+----------+
 

 Similarly, we can run group by and aggregate on two or more columns for other aggregate functions, please refer to the below example.

 4. Running more aggregates at a time

 Using agg() aggregate function we can calculate many aggregations at a time on a single statement using SQL functions sum(), avg(), min(), max() mean() e.t.c. In order to use these, we should import "from pyspark.sql.functions import sum,avg,max,min,mean,count"

 
 from pyspark.sql.functions import sum,avg,max
 df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus") \
 ) \
 .show(truncate=False)
 

 This example does group on department column and calculates sum() and avg() of salary for each department and calculates sum() and max() of bonus for each department.

 
 +----------+----------+-----------------+---------+---------+
 |department|sum_salary|avg_salary |sum_bonus|max_bonus|
 +----------+----------+-----------------+---------+---------+
 |Sales |257000 |85666.66666666667|53000 |23000 |
 |Finance |351000 |87750.0 |81000 |24000 |
 |Marketing |171000 |85500.0 |39000 |21000 |
 +----------+----------+-----------------+---------+---------+
 

 5. Using filter on aggregate data

 Similar to SQL “HAVING” clause, On PySpark DataFrame we can use either where() or filter() function to filter the rows of aggregated data.

 
 from pyspark.sql.functions import sum,avg,max
 df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus")) \
 .where(col("sum_bonus") >= 50000) \
 .show(truncate=False)
 

 This removes the sum of a bonus that has less than 50000 and yields below output.

 
 +----------+----------+-----------------+---------+---------+
 |department|sum_salary|avg_salary |sum_bonus|max_bonus|
 +----------+----------+-----------------+---------+---------+
 |Sales |257000 |85666.66666666667|53000 |23000 |
 |Finance |351000 |87750.0 |81000 |24000 |
 +----------+----------+-----------------+---------+---------+
 

 6. PySpark groupBy Example Source code

 
 import pyspark
 from pyspark.sql import SparkSession
 from pyspark.sql.functions import col,sum,avg,max
 
 spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
 
 simpleData = [("James","Sales","NY",90000,34,10000),
 ("Michael","Sales","NY",86000,56,20000),
 ("Robert","Sales","CA",81000,30,23000),
 ("Maria","Finance","CA",90000,24,23000),
 ("Raman","Finance","CA",99000,40,24000),
 ("Scott","Finance","NY",83000,36,19000),
 ("Jen","Finance","NY",79000,53,15000),
 ("Jeff","Marketing","CA",80000,25,18000),
 ("Kumar","Marketing","NY",91000,50,21000)
 ]
 
 schema = ["employee_name","department","state","salary","age","bonus"]
 df = spark.createDataFrame(data=simpleData, schema = schema)
 df.printSchema()
 df.show(truncate=False)
 
 df.groupBy("department").sum("salary").show(truncate=False)
 
 df.groupBy("department").count().show(truncate=False)
 
 
 df.groupBy("department","state") \
 .sum("salary","bonus") \
 .show(truncate=False)
 
 df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus") \
 ) \
 .show(truncate=False)
 
 df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus")) \
 .where(col("sum_bonus") >= 50000) \
 .show(truncate=False)
 

 This example is also available at GitHub PySpark Examples project for reference.

 7. Conclusion

 In this tutorial, you have learned how to use groupBy() functions on PySpark DataFrame and also learned how to run these on multiple columns and finally filter data on the aggregated columns.

 Thanks for reading. If you like it, please do share the article by following the below social links and any comments or suggestions are welcome in the comments sections!

 Happy Learning !!

 Related Articles

 * PySpark Groupby Count Distinct
 * PySpark Groupby on Multiple Columns
 * PySpark Groupby Agg (aggregate) – Explained
 * PySpark GroupBy Count – Explained
 * PySpark Select Top N Rows From Each Group
 * PySpark Find Maximum Row per Group in DataFrame
 * PySpark Select First Row of Each Group?
 * PySpark Column alias after groupBy() Example
 * PySpark DataFrame groupBy and Sort by Descending Order

 You may also like reading:

 1. Spark Groupby Example with DataFrame
 2. PySpark Pivot and Unpivot DataFrame
 3. PySpark withColumnRenamed to Rename Column on DataFrame
 4. What is Apache Spark Driver?
 5. PySpark GroupBy Count – Explained
 6. PySpark isNull() & isNotNull()
 7. Pyspark Select Distinct Rows
 8. PySpark Save DataFrame to Hive Table
 9. PySpark sum() Columns Example
 10. PySpark SQL Functions
 Tags: agg , groupby

 Leave a Reply

 Comment
 Enter your name or username to comment
 Enter your email address to comment
 Enter your website URL (optional)

 Save my name, email, and website in this browser for the next time I comment.

 Δ

 This Post Has 4 Comments

 1. Sneha October 31, 2020 Reply

 This is really great. I am learning pyspark in databricks and though there were a few syntax changes, the tutorial made me understand the concept properly.

 1. NNK October 31, 2020 Reply

 Thanks, Sneha for your comments, and glad you like the articles. If you find any syntax changes in Databricks please do comment, others might get benefit from your findings.

 2. Sr. Persona September 15, 2021 Reply

 Thanks, very useful

 3. Anonymous November 2, 2021 Reply

 It was very straightforward. Thank you very much.

 report this ad

 PySpark Tutorial

 * PySpark Tutorial For Beginners
 * PySpark – Features
 * PySpark – Advantages
 * PySpark – Modules & Packages
 * PySpark – Cluster Managers
 * PySpark – Install on Windows
 * PySpark – Install on Mac
 * PySpark – Web/Application UI
 * PySpark – SparkSession
 * PySpark – SparkContext
 * PySpark – RDD
 * PySpark – Parallelize
 * PySpark – repartition() vs coalesce()
 * PySpark – Broadcast Variables
 * PySpark – Accumulator

 PySpark DataFrame

 * PySpark – Create an empty DataFrame
 * PySpark – Convert RDD to DataFrame
 * PySpark – Convert DataFrame to Pandas
 * PySpark – show()
 * PySpark – StructType & StructField
 * PySpark – Column Class
 * PySpark – select()
 * PySpark – collect()
 * PySpark – withColumn()
 * PySpark – withColumnRenamed()
 * PySpark – where() & filter()
 * PySpark – drop() & dropDuplicates()
 * PySpark – orderBy() and sort()
 * PySpark – groupBy()
 * PySpark – join()
 * PySpark – union() & unionAll()
 * PySpark – unionByName()
 * PySpark – UDF (User Defined Function)
 * PySpark – transform()
 * PySpark – apply()
 * PySpark – map()
 * PySpark – flatMap()
 * PySpark – foreach()
 * PySpark – sample() vs sampleBy()
 * PySpark – fillna() & fill()
 * PySpark – pivot() (Row to Column)
 * PySpark – partitionBy()
 * PySpark – MapType (Map/Dict)

 PySpark SQL Functions

 * PySpark – Aggregate Functions
 * PySpark – Window Functions
 * PySpark – Date and Timestamp Functions
 * PySpark – JSON Functions

 PySpark Datasources

 * PySpark – Read & Write CSV File
 * PySpark – Read & Write Parquet File
 * PySpark – Read & Write JSON file
 * PySpark – Read Hive Table
 * PySpark – Save to Hive Table
 * PySpark – Read JDBC in Parallel
 * PySpark – Query Database Table
 * PySpark – Read and Write SQL Server
 * PySpark – Read and Write MySQL
 * PySpark – Read JDBC Table

 PySpark Built-in Functions

 * PySpark – when()
 * PySpark – expr()
 * PySpark – lit()
 * PySpark – split()
 * PySpark – concat_ws()
 * Pyspark – substring()
 * PySpark – translate()
 * PySpark – regexp_replace()
 * PySpark – overlay()
 * PySpark – to_timestamp()
 * PySpark – to_date()
 * PySpark – date_format()
 * PySpark – datediff()
 * PySpark – months_between()
 * PySpark – explode()
 * PySpark – array_contains()
 * PySpark – array()
 * PySpark – collect_list()
 * PySpark – collect_set()
 * PySpark – create_map()
 * PySpark – map_keys()
 * PySpark – map_values()
 * PySpark – struct()
 * PySpark – countDistinct()
 * PySpark – sum(), avg()
 * PySpark – row_number()
 * PySpark – rank()
 * PySpark – dense_rank()
 * PySpark – percent_rank()
 * PySpark – typedLit()
 * PySpark – from_json()
 * PySpark – to_json()
 * PySpark – json_tuple()
 * PySpark – get_json_object()
 * PySpark – schema_of_json()
 report this ad
 report this ad
 report this ad
 PySpark Tutorial For Beginners | Py...

 Please enable JavaScript

 PySpark Tutorial For Beginners | Python Examples
 report this ad
 report this ad
 Privacy & Cookies: This site uses cookies. By continuing to use this website, you agree to their use.
 To find out more, including how to control cookies, see here: Cookie Policy

 Top Tutorials

 * Apache Spark Tutorial
 * PySpark Tutorial
 * Python Pandas Tutorial
 * R Programming Tutorial
 * Python NumPy Tutorial
 * Apache Hive Tutorial
 * Apache HBase Tutorial
 * Apache Cassandra Tutorial
 * Apache Kafka Tutorial
 * Snowflake Data Warehouse Tutorial
 * H2O Sparkling Water Tutorial

 Categories

 * Apache Spark
 * PySpark
 * Pandas
 * R Programming
 * Snowflake Database
 * NumPy
 * Apache Hive
 * Apache HBase
 * Apache Kafka
 * Apache Cassandra
 * H2O Sparkling Water

 About SparkByExamples.com

 SparkByExamples.com is a Big Data and Spark examples community page, all examples are simple and easy to understand, and well tested in our development environment Read more ..
 * Opens in a new tab
 * Opens in a new tab
 * Opens in a new tab
 * Opens in a new tab
 * Opens in a new tab

 [email protected]
 +1 (949) 345-0676
 Desert Bloom
 Irvine , CA 92618
 USA
 Copyright sparkbyexamples.com
Similar to SQL GROUP BY clause, PySpark groupBy() function is used to collect the identical data into groups on DataFrame and perform count, sum, avg, min, max functions on the grouped data. In this article, I will explain several groupBy() examples using PySpark (Spark with Python). 
 Related: How to group and aggregate data using Spark and Scala 
1. GroupBy() Syntax &amp; Usage
 Syntax: 
 
# Syntax
DataFrame.groupBy(*cols)
#or 
DataFrame.groupby(*cols)
 
When we perform groupBy() on PySpark Dataframe, it returns GroupedData object which contains below aggregate functions.
 count() – Use groupBy() count() to return the number of rows for each group.
 mean() – Returns the mean of values for each group.
 max() – Returns the maximum of values for each group. 
 min() – Returns the minimum of values for each group.
 sum() – Returns the total for values for each group. 
 avg() – Returns the average for values for each group. 
 agg() – Using groupBy() agg() function, we can calculate more than one aggregate at a time.
 pivot() – This function is used to Pivot the DataFrame which I will not be covered in this article as I already have a dedicated article for Pivot &amp; Unpivot DataFrame.
Before we start, let’s create the DataFrame from a sequence of the data to work with. This DataFrame contains columns “employee_name”, “department”, “state“, “salary”, “age” and “bonus” columns.
We will use this PySpark DataFrame to run groupBy() on “department” columns and calculate aggregates like minimum, maximum, average, and total salary for each group using min(), max(), and sum() aggregate functions respectively.
 
simpleData = [("James","Sales","NY",90000,34,10000),
 ("Michael","Sales","NY",86000,56,20000),
 ("Robert","Sales","CA",81000,30,23000),
 ("Maria","Finance","CA",90000,24,23000),
 ("Raman","Finance","CA",99000,40,24000),
 ("Scott","Finance","NY",83000,36,19000),
 ("Jen","Finance","NY",79000,53,15000),
 ("Jeff","Marketing","CA",80000,25,18000),
 ("Kumar","Marketing","NY",91000,50,21000)
 ]
schema = ["employee_name","department","state","salary","age","bonus"]
df = spark.createDataFrame(data=simpleData, schema = schema)
df.printSchema()
df.show(truncate=False)
 
Yields below output.  
+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
| James| Sales| NY| 90000| 34|10000|
| Michael| Sales| NY| 86000| 56|20000|
| Robert| Sales| CA| 81000| 30|23000|
| Maria| Finance| CA| 90000| 24|23000|
| Raman| Finance| CA| 99000| 40|24000|
| Scott| Finance| NY| 83000| 36|19000|
| Jen| Finance| NY| 79000| 53|15000|
| Jeff| Marketing| CA| 80000| 25|18000|
| Kumar| Marketing| NY| 91000| 50|21000|
+-------------+----------+-----+------+---+-----+
 
2. PySpark groupBy on DataFrame Columns
Let’s do the groupBy() on department column of DataFrame and then find the sum of salary for each department using sum() function.
 
df.groupBy("department").sum("salary").show(truncate=False)
+----------+-----------+
|department|sum(salary)|
+----------+-----------+
|Sales |257000 |
|Finance |351000 |
|Marketing |171000 |
+----------+-----------+
 
Similarly, we can calculate the number of employees in each department using.
 
df.groupBy("department").count()
 
Calculate the minimum salary of each department using min() 
 
df.groupBy("department").min("salary")
 
Calculate the maximin salary of each department using max()   
df.groupBy("department").max("salary")
 
Calculate the average salary of each department using avg() 
 
df.groupBy("department").avg( "salary")
 
Calculate the mean salary of each department using mean() 
 
df.groupBy("department").mean( "salary") 
 
3. Using Multiple columns 
Similarly, we can also run groupBy and aggregate on two or more DataFrame columns, below example does group by on department,state and does sum() on salary and bonus columns.  
#GroupBy on multiple columns
df.groupBy("department","state") \
 .sum("salary","bonus") \
 .show(false)
 
This yields the below output.
 
+----------+-----+-----------+----------+
|department|state|sum(salary)|sum(bonus)|
+----------+-----+-----------+----------+
|Finance |NY |162000 |34000 |
|Marketing |NY |91000 |21000 |
|Sales |CA |81000 |23000 |
|Marketing |CA |80000 |18000 |
|Finance |CA |189000 |47000 |
|Sales |NY |176000 |30000 |
+----------+-----+-----------+----------+
 
Similarly, we can run group by and aggregate on two or more columns for other aggregate functions, please refer to the below example.
4. Running more aggregates at a time
Using agg() aggregate function we can calculate many aggregations at a time on a single statement using SQL functions sum(), avg(), min(), max() mean() e.t.c. In order to use these, we should import "from pyspark.sql.functions import sum,avg,max,min,mean,count" 
 
from pyspark.sql.functions import sum,avg,max
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus") \
 ) \
 .show(truncate=False)
 
This example does group on department column and calculates sum() and avg() of salary for each department and calculates sum() and max() of bonus for each department.
 
+----------+----------+-----------------+---------+---------+
|department|sum_salary|avg_salary |sum_bonus|max_bonus|
+----------+----------+-----------------+---------+---------+
|Sales |257000 |85666.66666666667|53000 |23000 |
|Finance |351000 |87750.0 |81000 |24000 |
|Marketing |171000 |85500.0 |39000 |21000 |
+----------+----------+-----------------+---------+---------+
 
5. Using filter on aggregate data
Similar to SQL “HAVING” clause, On PySpark DataFrame we can use either where() or filter() function to filter the rows of aggregated data.
 
from pyspark.sql.functions import sum,avg,max
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus")) \
 .where(col("sum_bonus") &gt;= 50000) \
 .show(truncate=False)
 
This removes the sum of a bonus that has less than 50000 and yields below output.
 
+----------+----------+-----------------+---------+---------+
|department|sum_salary|avg_salary |sum_bonus|max_bonus|
+----------+----------+-----------------+---------+---------+
|Sales |257000 |85666.66666666667|53000 |23000 |
|Finance |351000 |87750.0 |81000 |24000 |
+----------+----------+-----------------+---------+---------+
 
6. PySpark groupBy Example Source code
 
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col,sum,avg,max
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
simpleData = [("James","Sales","NY",90000,34,10000),
 ("Michael","Sales","NY",86000,56,20000),
 ("Robert","Sales","CA",81000,30,23000),
 ("Maria","Finance","CA",90000,24,23000),
 ("Raman","Finance","CA",99000,40,24000),
 ("Scott","Finance","NY",83000,36,19000),
 ("Jen","Finance","NY",79000,53,15000),
 ("Jeff","Marketing","CA",80000,25,18000),
 ("Kumar","Marketing","NY",91000,50,21000)
 ]
schema = ["employee_name","department","state","salary","age","bonus"]
df = spark.createDataFrame(data=simpleData, schema = schema)
df.printSchema()
df.show(truncate=False)
df.groupBy("department").sum("salary").show(truncate=False)
df.groupBy("department").count().show(truncate=False)
df.groupBy("department","state") \
 .sum("salary","bonus") \
 .show(truncate=False)
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus") \
 ) \
 .show(truncate=False)
 
df.groupBy("department") \
 .agg(sum("salary").alias("sum_salary"), \
 avg("salary").alias("avg_salary"), \
 sum("bonus").alias("sum_bonus"), \
 max("bonus").alias("max_bonus")) \
 .where(col("sum_bonus") &gt;= 50000) \
 .show(truncate=False)
 
This example is also available at GitHub PySpark Examples project for reference.
7. Conclusion
In this tutorial, you have learned how to use groupBy() functions on PySpark DataFrame and also learned how to run these on multiple columns and finally filter data on the aggregated columns.
Thanks for reading. If you like it, please do share the article by following the below social links and any comments or suggestions are welcome in the comments sections!  
Happy Learning !!
Related Articles
 PySpark Groupby Count Distinct 
 PySpark Groupby on Multiple Columns 
 PySpark Groupby Agg (aggregate) – Explained 
 PySpark GroupBy Count – Explained 
 PySpark Select Top N Rows From Each Group 
 PySpark Find Maximum Row per Group in DataFrame 
 PySpark Select First Row of Each Group? 
 PySpark Column alias after groupBy() Example 
 PySpark DataFrame groupBy and Sort by Descending Order
PySpark Groupby Explained with Example - Spark By {Examples}
https://sparkbyexamples.com/pyspark/pyspark-groupby-explained-with-example/