How to select the first row of each group

Navigating complex datasets often requires extracting very specific information, and a common challenge data professionals face is figuring out how to select the first row of each group. Whether you’re dealing with sales transactions, user activity logs, or sensor data, identifying the initial record within a defined group is a fundamental task. This isn’t just about picking an arbitrary row; it typically involves a logical ordering to determine what “first” truly means. Mastering this technique is crucial for data cleaning, reporting, and analytical tasks, ensuring you capture the most relevant starting point for each category in your dataset. We’ll explore robust SQL methods to tackle this, moving from powerful window functions to more traditional subquery approaches, equipping you with the knowledge to efficiently manage your data.

Understanding the “First Row Per Group” Challenge in Data Analysis

The concept of selecting the “first row of each group” might seem straightforward, but its implementation requires careful consideration of what defines both “first” and “group.” A “group” refers to a subset of rows that share a common characteristic, such as all transactions by a specific customer, all sensor readings from a particular device, or all versions of a product. The “first” row within that group is usually determined by an ordering criterion, like the earliest timestamp, the lowest ID number, or a specific status priority.

This challenge frequently arises when you need to de-duplicate data, retrieve the most current or initial state of an entity, or simplify complex historical records. For instance, if you have multiple entries for a user’s address changes over time, you might only want their very first recorded address. Or, in a log of product updates, you might need to identify the initial version released for each product. Without a precise method, you risk either returning too many rows or, worse, incorrect data that could skew your analysis.

Traditional SQL clauses like GROUP BY are excellent for aggregation (e.g., summing sales per product), but they don’t directly facilitate selecting an entire row based on an ordering within that group. This limitation necessitates more advanced SQL techniques that can partition data, apply an order, and then pick the desired row. Understanding these underlying principles is the first step towards effectively manipulating your grouped data.

Leveraging Window Functions: ROW_NUMBER() and PARTITION BY

When it comes to efficiently selecting the first row of each group, SQL window functions, particularly ROW_NUMBER() combined with PARTITION BY, are often the most powerful and flexible solution. This method allows you to assign a unique, sequential integer to each row within its partition (group), based on a specified order. Once ranked, it becomes trivial to filter for the row with rank 1.

What is ROW_NUMBER() and PARTITION BY?

ROW_NUMBER() is a ranking window function that assigns a unique number to each row within a partition, based on the specified order. The numbering starts from 1 for the first row in each partition. PARTITION BY divides the result set into partitions to which the ROW_NUMBER() function is applied independently. Think of it as creating temporary, distinct groups of rows. The ORDER BY clause within the OVER() specification determines what “first” means for each group.

For example, if you have a table of product versions and you want the earliest version for each product, you would partition by ProductID and order by VersionDate. This approach is highly efficient because it processes the data in a single pass, avoiding multiple table scans or complex joins that might occur with subquery-based methods. According to a study by Redgate, optimizing queries with window functions can lead to significant performance improvements, especially on large datasets. Source: Redgate Simple Talk.

Step-by-Step Implementation with ROW_NUMBER()

Here’s how you typically implement this using a Common Table Expression (CTE) for clarity and efficiency:

Define Your Grouping Column(s): Identify the column(s) that define your groups (e.g., CustomerID, ProductID). These will go into the PARTITION BY clause.
Define Your Ordering Column(s): Determine what makes a row “first” within each group (e.g., TransactionDate, RecordID). These will go into the ORDER BY clause within the OVER() specification.
Apply ROW_NUMBER(): Create a CTE or subquery that uses ROW_NUMBER() OVER (PARTITION BY [GroupingColumn(s)] ORDER BY [OrderingColumn(s)] [ASC/DESC]) AS rn.
Filter for Rank 1: Select from your CTE/subquery where the assigned rank (rn) is equal to 1.

Example SQL:

WITH RankedTransactions AS ( SELECT TransactionID, CustomerID, TransactionDate, Amount, ROW_NUMBER() OVER (PARTITION BY CustomerID ORDER BY TransactionDate ASC) AS rn FROM Sales.Transactions ) SELECT TransactionID, CustomerID, TransactionDate, Amount FROM RankedTransactions WHERE rn = 1;

This method ensures that for every unique CustomerID, you retrieve only the transaction with the earliest TransactionDate. This is incredibly useful for analyzing initial customer behavior or their first purchase records.

Simplifying with DISTINCT ON (PostgreSQL) -----------------------------------------

For users working specifically with PostgreSQL databases, there’s a highly efficient and concise alternative to the ROW_NUMBER() method for selecting the first row of each group: the DISTINCT ON clause. This feature is a powerful extension to the standard SQL SELECT DISTINCT statement.

The DISTINCT ON clause processes the result set and, for each set of rows where the expressions specified in DISTINCT ON are equal, it keeps only the “first” row. What determines “first” is crucial here: it’s dictated by the ORDER BY clause of the main query. This means you must carefully construct your ORDER BY to achieve the desired “first” row within each group.

For instance, if you want the most recent order for each customer, you would use DISTINCT ON (CustomerID) and then ORDER BY CustomerID, OrderDate DESC. PostgreSQL will group by CustomerID, and for each group, it will pick the first row based on the descending OrderDate. This elegant syntax often results in cleaner and more readable queries compared to CTEs with window functions, especially for simpler “first row” scenarios.

**Advantages of DISTINCT Question & Answer :
I have a DataFrame generated as follow:

df.groupBy($"Hour", $"Category") .agg(sum($"value") as "TotalValue") .sort($"Hour".asc, $"TotalValue".desc))

The results look like:

+----+--------+----------+ |Hour|Category|TotalValue| +----+--------+----------+ | 0| cat26| 30.9| | 0| cat13| 22.1| | 0| cat95| 19.6| | 0| cat105| 1.3| | 1| cat67| 28.5| | 1| cat4| 26.8| | 1| cat13| 12.6| | 1| cat23| 5.3| | 2| cat56| 39.6| | 2| cat40| 29.7| | 2| cat187| 27.9| | 2| cat68| 9.8| | 3| cat8| 35.6| | ...| ....| ....| +----+--------+----------+

As you can see, the DataFrame is ordered by Hour in an increasing order, then by TotalValue in a descending order.

I would like to select the top row of each group, i.e.

- from the group of Hour==0 select (0,cat26,30.9) - from the group of Hour==1 select (1,cat67,28.5) - from the group of Hour==2 select (2,cat56,39.6) - and so on

So the desired output would be:

+----+--------+----------+ |Hour|Category|TotalValue| +----+--------+----------+ | 0| cat26| 30.9| | 1| cat67| 28.5| | 2| cat56| 39.6| | 3| cat8| 35.6| | ...| ...| ...| +----+--------+----------+

It might be handy to be able to select the top N rows of each group as well.

Any help is highly appreciated.

Window functions:

Something like this should do the trick:

import org.apache.spark.sql.functions.{row_number, max, broadcast} import org.apache.spark.sql.expressions.Window val df = sc.parallelize(Seq( (0,"cat26",30.9), (0,"cat13",22.1), (0,"cat95",19.6), (0,"cat105",1.3), (1,"cat67",28.5), (1,"cat4",26.8), (1,"cat13",12.6), (1,"cat23",5.3), (2,"cat56",39.6), (2,"cat40",29.7), (2,"cat187",27.9), (2,"cat68",9.8), (3,"cat8",35.6))).toDF("Hour", "Category", "TotalValue") val w = Window.partitionBy($"hour").orderBy($"TotalValue".desc) val dfTop = df.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop("rn") dfTop.show // +----+--------+----------+ // |Hour|Category|TotalValue| // +----+--------+----------+ // | 0| cat26| 30.9| // | 1| cat67| 28.5| // | 2| cat56| 39.6| // | 3| cat8| 35.6| // +----+--------+----------+

This method will be inefficient in case of significant data skew. This problem is tracked by SPARK-34775 and might be resolved in the future (SPARK-37099).

Plain SQL aggregation followed by join:

Alternatively you can join with aggregated data frame:

val dfMax = df.groupBy($"hour".as("max_hour")).agg(max($"TotalValue").as("max_value")) val dfTopByJoin = df.join(broadcast(dfMax), ($"hour" === $"max_hour") && ($"TotalValue" === $"max_value")) .drop("max_hour") .drop("max_value") dfTopByJoin.show // +----+--------+----------+ // |Hour|Category|TotalValue| // +----+--------+----------+ // | 0| cat26| 30.9| // | 1| cat67| 28.5| // | 2| cat56| 39.6| // | 3| cat8| 35.6| // +----+--------+----------+

It will keep duplicate values (if there is more than one category per hour with the same total value). You can remove these as follows:

dfTopByJoin .groupBy($"hour") .agg( first("category").alias("category"), first("TotalValue").alias("TotalValue"))

Using ordering over structs:

Neat, although not very well tested, trick which doesn’t require joins or window functions:

val dfTop = df.select($"Hour", struct($"TotalValue", $"Category").alias("vs")) .groupBy($"hour") .agg(max("vs").alias("vs")) .select($"Hour", $"vs.Category", $"vs.TotalValue") dfTop.show // +----+--------+----------+ // |Hour|Category|TotalValue| // +----+--------+----------+ // | 0| cat26| 30.9| // | 1| cat67| 28.5| // | 2| cat56| 39.6| // | 3| cat8| 35.6| // +----+--------+----------+

With DataSet API (Spark 1.6+, 2.0+):

Spark 1.6:

case class Record(Hour: Integer, Category: String, TotalValue: Double) df.as[Record] .groupBy($"hour") .reduce((x, y) => if (x.TotalValue > y.TotalValue) x else y) .show // +---+--------------+ // | _1| _2| // +---+--------------+ // |[0]|[0,cat26,30.9]| // |[1]|[1,cat67,28.5]| // |[2]|[2,cat56,39.6]| // |[3]| [3,cat8,35.6]| // +---+--------------+

Spark 2.0 or later:

df.as[Record] .groupByKey(_.Hour) .reduceGroups((x, y) => if (x.TotalValue > y.TotalValue) x else y)

The last two methods can leverage map side combine and don’t require full shuffle so most of the time should exhibit a better performance compared to window functions and joins. These cane be also used with Structured Streaming in completed output mode.

Don’t use:

df.orderBy(...).groupBy(...).agg(first(...), ...)

It may seem to work (especially in the local mode) but it is unreliable (see SPARK-16207, credits to Tzach Zohar for linking relevant JIRA issue, and SPARK-30335).

The same note applies to

df.orderBy(...).dropDuplicates(...)

which internally uses equivalent execution plan.**