DataSet Operations – Aggregation – Window Functions

Before we dive headfirst into Window functions it would be a good idea to review the groupby aggregate functions. An important thing to note, when groupBy functions are applied they reduce the number of rows returned.

However, when Window functions are applied they do not reduce the rows in the result set. Window describes the set of rows on which the window function operates. It returns a value which is shown in all the rows of the window. One more thing, Window Functions also help us compare the current rows with other rows.

Simple Aggregations

Below is an example showing two operations to give a contrast in the results.

• avg function when a groupBy is applied
• avg function when a window is applied

Let’s look at the Step – 4 & 5 which are the most important in this example.

• Step 1,2,3 – Creating a spark context, Reading a data file
• Step 3 – Doing a groupBy average
• Step 4 – Create a Window expression- also called the Window Spec – which is telling spark to create sets of rows based on the values of Product Type column.
• Step 5 – Applies the Window Spec over an average aggregation.

Observe that there is no groupBy method call when the window function is applied.

Let’s look at some more interesting examples like applying a rank & dense_rank.

The results from the log file are shown below.

Observe that all the rows are returned and the average quantity (avg_Quantity) is generated for every Product Type(ProductType). Each product type has a same the value for average quantity.

Rank & Dense Rank

Let’s look at another interesting example of ranking the data based on its value in a Window. So, we want to rank the quantity in the decrease order. We will use two functions rank and dense_rank.

• The rank function assigns ranking within an ordered window. If the values of the two rows are the same, the rank function assigns the same rank, with the next ranking(s) skipped.
• the dense_rank function assigns the ranking within an ordered window, but the ranks are consecutive. In other words, the same ranks are assigned to multiple rows and no ranks are skipped when compared to rank.

Let’s look at an example showing both the functions together.

The interesting part of this program is in Step3 & Step4

• Step 1,2 – Create a spark session and read the data file.
• Step 3 – Create a Window expression- also called the Window Spec – which is telling spark to create sets of rows based on the values of Product Type column. In addition to that, we use orderBy function and ask spark to order the rows based on values in the quantity column.
• Step 4 – Applies the Window Spec over an average aggregation.
• Step 5 – Show the result.

The relevant output from the data is shown below

Observe how the ranks are created for rankForQty and denseRankForQty. When the rank function is applied – values which have the same value are assigned the same rank. Additionally, the subsequent ranks are skipped as is the case where quantity is 100,100,101 – the rank is 1,1,3 – rank 2 is skipped. dense_rank does not skip the rank so for the same example where quantity is 100,100,101 – the dense rank is 1,1,2 and no rank is skipped.

Cumulative Aggregations

Turning a page let’s now take another use case which comes along a lot. Having cumulative aggregations. A very commonly needed use case is a cumulative or running total. Let’s take look at how we can create a cumulative aggregate.

See the example below with

Here we introduce a new method rowsBetween which allows us to set create cumulative aggregations. This method is passed two parameters it allows us to define a Frame with respect to the current row.

• Start – In a partition, from where does the frame start with respect to the current row. In cumulative sum, the parameter is Window.unboundedPreceding.
• End – In a partition, where does the frame end with respect to the current row. In cumulative sum, the parameter is Window.currentRow

Essentially, we are telling spark to go all the way to the start of the Window and from there look till the current row.

The output of the relevant part of the log is shown below. Observe the cumulativeQty with respective to the Quantity column.
cumulativeQty is running total of Quantity Column

Lead & Lag Functions

The next set of operators which allows us to compare the current row with the next row(s) or the previous row(s) in a Window. This is done by providing an offset to compare. Offset is relative to the current row.

For example, if we want to compare the quantity of the current row with the quantity in the next row we just say lead(“quantity”,1). Similarly, if you want to compare the quantity with the second row after the current row we just say lead(“quantity”,2). The lag function works in exactly the same way but in the other direction.

See below an example showing lead and lag functions in action.

Take a look at the Step-4 closely and it should be self-explanatory by now. In case there is no value found by the lead or the lag method, because the offset is less than the number of rows which are there a null is populated. If you want some other default value can also be populated and can be mentioned as a part of the method call. See the method signature for lead below. These are available in org.apache.spark.sql.functions

The relevant part of the output is below

First and Last Value

Finally, we come to the last two methods in the Window functions and they give is the first value and the last value of a column in a Window. They are pretty simple to implement. See the code below.

Essentially, we are telling spark to start from the first row of the Window and go all the go all the way to the end of the Window while it is looking for the first and the last value.

The relevant log is show below

If you are interested in learning more – I advise you to start looking at the spark documentation here. and look for Window functions

A sample of data for this blog is given below. It is a subset of data published by IBM for Watson Analytics

Hope you have found this blog entry useful.

Till next time…..Byeee!