GroupBy and aggregations in Apache Spark

Introduction

GroupBy helps you organize data into groups based on one or more columns. Aggregations let you calculate summary values like sums or averages for each group.

You want to find total sales per product category.

You need to calculate average temperature per city.

You want to count how many users belong to each country.

You want to find the maximum score per student in a test.

Syntax

Apache Spark

df.groupBy("column_name").agg(functions)

Use groupBy to split data into groups by column(s).

Use agg to apply aggregation functions like sum, avg, count.

Examples

Groups data by 'category' and sums the 'sales' in each group.

Apache Spark

df.groupBy("category").sum("sales")

Groups data by 'city' and calculates average temperature per city.

Apache Spark

df.groupBy("city").agg({'temperature': 'avg'})

Groups data by 'country' and counts rows in each group.

Apache Spark

df.groupBy("country").count()

Sample Program

This program creates a Spark DataFrame with sales data by category. It then groups the data by category and calculates the sum, average, and maximum sales for each group. Finally, it prints the results.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum, avg, max

spark = SparkSession.builder.appName("GroupByExample").getOrCreate()

# Sample data
data = [
    ("Electronics", 100),
    ("Electronics", 150),
    ("Clothing", 200),
    ("Clothing", 50),
    ("Groceries", 300)
]

# Create DataFrame
columns = ["category", "sales"]
df = spark.createDataFrame(data, columns)

# Group by category and sum sales
sum_sales = df.groupBy("category").sum("sales")

# Group by category and average sales
avg_sales = df.groupBy("category").agg(avg("sales").alias("avg_sales"))

# Group by category and max sales
max_sales = df.groupBy("category").agg(max("sales").alias("max_sales"))

# Show results
print("Sum of sales by category:")
sum_sales.show()

print("Average sales by category:")
avg_sales.show()

print("Max sales by category:")
max_sales.show()

spark.stop()

Important Notes

You can group by multiple columns by passing a list to groupBy, like groupBy(['col1', 'col2']).

Aggregation functions like sum, avg, max, min, and count are available in pyspark.sql.functions.

Always give meaningful names to aggregated columns using alias for clearer output.

Summary

GroupBy splits data into groups based on column values.

Aggregations calculate summary statistics for each group.

Use Spark functions like sum, avg, max inside agg for flexible summaries.

Practice

1. What does the groupBy function do in Apache Spark?

easy

A. It splits the data into groups based on column values.

B. It sorts the data in ascending order.

C. It filters rows based on a condition.

D. It joins two DataFrames together.

Solution

1. Step 1: Understand the purpose of groupBy

   The groupBy function is used to split data into groups based on one or more column values.

2. Step 2: Compare with other functions

   Sorting, filtering, and joining are different operations and do not split data into groups.

3. Final Answer:

   It splits the data into groups based on column values. -> Option A

4. Quick Check:

   groupBy = splits data into groups [OK]

Hint: Remember: groupBy groups rows by column values [OK]

Common Mistakes:

• Confusing groupBy with filter or sort
• Thinking groupBy changes row order
• Assuming groupBy merges DataFrames

2. Which of the following is the correct syntax to calculate the average of column sales after grouping by region in Spark DataFrame df?

easy

A. df.groupBy("region").sum("sales")

B. df.groupBy("region").agg(avg("sales"))

C. df.groupBy("region").filter(avg("sales"))

D. df.groupBy("region").select(avg("sales"))

Solution

1. Step 1: Identify correct aggregation syntax

   To calculate average after grouping, use groupBy followed by agg(avg(column)).

2. Step 2: Check each option

   df.groupBy("region").agg(avg("sales")) uses agg(avg("sales")) correctly. df.groupBy("region").sum("sales") uses sum, not average. df.groupBy("region").filter(avg("sales")) uses filter incorrectly. df.groupBy("region").select(avg("sales")) uses select which is not valid after groupBy for aggregation.

3. Final Answer:

   df.groupBy("region").agg(avg("sales")) -> Option B

4. Quick Check:

   groupBy + agg(avg) = df.groupBy("region").agg(avg("sales")) [OK]

Hint: Use agg() with avg() inside groupBy for averages [OK]

Common Mistakes:

• Using sum instead of avg for average
• Using filter instead of agg
• Using select after groupBy without agg

3. Given the following Spark code, what will be the output schema and result?

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum

spark = SparkSession.builder.getOrCreate()
data = [("East", 100), ("West", 200), ("East", 150)]
df = spark.createDataFrame(data, ["region", "sales"])
result = df.groupBy("region").agg(sum("sales").alias("total_sales"))
result.show()

medium

A. region | total_sales
East | 250
West | 200

B. region | sales
East | 100
West | 200
East | 150

C. region | total_sales
East | 150
West | 200

D. region | total_sales
East | 100
West | 200

Solution

1. Step 1: Understand groupBy and sum aggregation

   The code groups rows by "region" and sums the "sales" values for each group.

2. Step 2: Calculate sums for each region

   For "East": 100 + 150 = 250; For "West": 200.

3. Final Answer:

   region | total_sales East | 250 West | 200 -> Option A

4. Quick Check:

   Sum sales by region = 250 (East), 200 (West) [OK]

Hint: Sum sales per group to get total_sales [OK]

Common Mistakes:

• Not summing all values per group
• Confusing alias name with original column
• Expecting original rows without aggregation

4. Identify the error in the following Spark code snippet:

df.groupBy("category").agg(sum("amount"), avg("amount"))

medium

A. Missing alias for aggregated columns causes error.

B. agg() requires a dictionary, not multiple arguments.

C. groupBy cannot be used with agg.

D. sum and avg functions are not imported.

Solution

1. Step 1: Check function imports

   The code uses sum and avg functions but does not show imports from pyspark.sql.functions.

2. Step 2: Validate other options

   agg() accepts multiple arguments; aliasing is optional; groupBy works with agg.

3. Final Answer:

   sum and avg functions are not imported. -> Option D

4. Quick Check:

   Missing imports cause NameError [OK]

Hint: Always import aggregation functions before use [OK]

Common Mistakes:

• Assuming agg needs dict, but it accepts multiple args
• Forgetting to import sum, avg from pyspark.sql.functions
• Thinking groupBy can't be chained with agg

5. You have a Spark DataFrame sales_df with columns store, product, and revenue. You want to find the maximum revenue per store and the average revenue per product across all stores in a single query. Which code snippet correctly achieves this?

hard

A. sales_df.groupBy("store").agg(max("revenue").alias("max_revenue"), avg("revenue").alias("avg_revenue"))

B. sales_df.agg(max("revenue").alias("max_revenue"), avg("revenue").alias("avg_revenue"))

C. sales_df.groupBy("store").agg(max("revenue").alias("max_revenue")).crossJoin(sales_df.groupBy("product").agg(avg("revenue").alias("avg_revenue")))

D. sales_df.groupBy("store").agg(max("revenue").alias("max_revenue")).groupBy("product").agg(avg("revenue").alias("avg_revenue"))

Solution

1. Step 1: Understand the requirements

   We need max revenue per store and average revenue per product in one result.

2. Step 2: Analyze options

   sales_df.groupBy("store").agg(max("revenue").alias("max_revenue"), avg("revenue").alias("avg_revenue")) aggregates both by store only, so avg revenue per product is incorrect. sales_df.agg(max("revenue").alias("max_revenue"), avg("revenue").alias("avg_revenue")) aggregates globally, losing grouping. sales_df.groupBy("store").agg(max("revenue").alias("max_revenue")).groupBy("product").agg(avg("revenue").alias("avg_revenue")) tries chaining groupBy but overwrites previous aggregation. sales_df.groupBy("store").agg(max("revenue").alias("max_revenue")).crossJoin(sales_df.groupBy("product").agg(avg("revenue").alias("avg_revenue"))) computes both groupings separately and joins results, correctly combining both summaries.

3. Final Answer:

   sales_df.groupBy("store").agg(max("revenue").alias("max_revenue")).crossJoin(sales_df.groupBy("product").agg(avg("revenue").alias("avg_revenue"))) -> Option C

4. Quick Check:

   Separate groupBy + crossJoin combines both aggregations [OK]

Hint: Use separate groupBy then join to combine different aggregations [OK]

Common Mistakes:

• Trying to aggregate different groupings in one groupBy
• Overwriting aggregation by chaining groupBy
• Forgetting to join separate aggregation results