Introduction
When processing data we often find ourselves in a situation where we want to calculate
variables over certain subset of observations. For example, we might be interested in the
average value per group or the maximum value for each group. The groupBy function
available in many programming or consulting programs allows us to do these calculations
easily, simply by specifying the variables that define the final group and the functions that we
want to apply within each group.
Normally after applying groupBy, only one observation is obtained for each of the groups are
defined, which means that if we wanted to compare the aggregates obtained with the
observations that generated them, we would have to create a specific function or use a join
of the original table with the results of the groupBy.
To facilitate tasks like this, many programming languages have the Window function. This
has a similar function to groupBy with the difference that Window does not modify the
number of rows of the DataFrame. In this post we will explain how to use Window in Apache
Spark, specifically in its implementation in pySpark.
To compare the behaviour of groupBy with Window, let’s imagine the following problem: We
have a set of students and for each one we have the class they were in and the grade
obtained. From this data, we want to determine if a student got a grade higher than the
average of his class.
Let’s generate some simulated data for testing:
~~~~ python
grades = [('ES001', 'English', 18),
('ES002', 'English', 14),
('ES003', 'English', 19),
('ES001', 'French', 13),
('ES002', 'French', 16),
('ES003', 'French', 11),
('ES001', 'Maths', 15),
('ES002', 'Maths', 17),
('ES003', 'Maths', 19)]
gradesSpark = spark.createDataFrame(grades, schema=['student', 'course', 'grade']).cache()
gradesSpark.show()
~~~~
~~~~ console
+-------+-------+-----+
|student| course|grade|
+-------+-------+-----+
| ES001|English| 18|
| ES002|English| 14|
| ES003|English| 19|
| ES001| French| 13|
| ES002| French| 16|
| ES003| French| 11|
| ES001| Maths| 15|
| ES002| Maths| 17|
| ES003| Maths| 19|
+-------+-------+-----+
~~~~To solve this problem using only groupBy we first group by class to obtain the average and
then to match this variable with the original data we use a join:
~~~~ python
result = gradesSpark.join(gradesSpark.groupBy('course').agg(avg(col('grade')).alias('courseAvg')),
on=['course'],
how='left')
result = result.withColumn('aboveAvg', when(col('grade')>col('courseAvg'), lit(1)).otherwise(lit(0)))
result.show()
~~~~
~~~~ console
+-------+-------+-----+------------------+--------+
| course|student|grade| courseAvg|aboveAvg|
+-------+-------+-----+------------------+--------+
|English| ES001| 18| 17.0| 1|
|English| ES002| 14| 17.0| 0|
|English| ES003| 19| 17.0| 1|
| French| ES001| 13|13.333333333333334| 0|
| French| ES002| 16|13.333333333333334| 1|
| French| ES003| 11|13.333333333333334| 0|
| Maths| ES001| 15| 17.0| 0|
| Maths| ES002| 17| 17.0| 0|
| Maths| ES003| 19| 17.0| 1|
+-------+-------+-----+------------------+--------+
~~~~This same example using Window is much simpler since we only have to define the window
in which we want to apply the function and then create the variable using the over() function
indicating the window we want to use for the calculation:
~~~~ python
win = Window.partitionBy('course')
result = gradesSpark.withColumn('courseAvg', avg(col('grade')).over(win))
result = result.withColumn('aboveAvg', when(col('grade')>col('courseAvg'), lit(1)).otherwise(lit(0)))
result.show()
~~~~
~~~~ console
+-------+-------+-----+------------------+--------+
|student| course|grade| courseAvg|aboveAvg|
+-------+-------+-----+------------------+--------+
| ES003|English| 19| 17.0| 1|
| ES001|English| 18| 17.0| 1|
| ES002|English| 14| 17.0| 0|
| ES001| French| 13|13.333333333333334| 0|
| ES002| French| 16|13.333333333333334| 1|
| ES003| French| 11|13.333333333333334| 0|
| ES001| Maths| 15| 17.0| 0|
| ES002| Maths| 17| 17.0| 0|
| ES003| Maths| 19| 17.0| 1|
+-------+-------+-----+------------------+--------+
~~~~Window coding is much less complicated and more efficient since there is no need to do a
join which can often be very computationally expensive.
Defining windows
Spark provides us with different functions that we can combine to sit many window types
easily:
partitionBy
This function is the one we used in the previous example and serves to define the groups
that we have inside the data. It needs as an argument a list with the variables that delimit
such groups as when we use a groupBy.
orderBy
It allows us to organise the data within a group. In addition, when using orderBy alone or
with partitionBy when defining the window, the calculations are performed for all rows whose
value precedes or is equal to that of the current row. This way, if two rows have the same
value for the variable being ordered, both will enter the calculation.
To exemplify the use of orderBy let’s use data on the contribution of different sectors to PIB
and imagine that we want to keep the largest sectors whose combined contribution is at
least 80%. For this we can use a window with orderBy to define a descending order
according to the percentage of PIB and sum up:
~~~~ python
gdp = [('Agriculture', 5),
('Telecom', 25),
('Tourism', 40),
('Petrochemical', 25),
('Construction', 5)]
gdpSpark = spark.createDataFrame(gdp, schema=['sector', 'percGdp'])
win = Window().orderBy(col('percGdp').desc())
win2 = Window().orderBy(col('percGdp').desc()).rowsBetween(Window.unboundedPreceding, Window.currentRow)
gdpSpark.withColumn('percBiggerSectors', sum(col('percGdp')).over(win)).show()
~~~~
~~~~ console
+-------------+-------+-----------------+
| sector|percGdp|percBiggerSectors|
+-------------+-------+-----------------+
| Tourism| 40| 40|
| Telecom| 25| 90|
|Petrochemical| 25| 90|
| Construction| 5| 100|
| Agriculture| 5| 100|
+-------------+-------+-----------------+
~~~~Note that for Telecommunications the value of contributions MasGrandes is 90=40+25+25,
since orderBy does not take into account that Petroquímica is in the next row, but that the
value 25 is the same in both rows. Another important feature that emerges from this example
is that it is not necessary to use partitionBy together with orderBy. If partionBy is not
included, in the group is simply all data.
orderBy is also useful when we want to use order to create variables. For example, with the
grade data we used in the previous section we could create a ranking of students according
to their grade within each class:
~~~~ python
win = Window.partitionBy('course').orderBy(col('grade').desc())
result = gradesSpark.withColumn('ranking', rank().over(win))
result.show()
~~~~
~~~~ console
+-------+-------+-----+-------+
|student| course|grade|ranking|
+-------+-------+-----+-------+
| ES003|English| 19| 1|
| ES001|English| 18| 2|
| ES002|English| 14| 3|
| ES002| French| 16| 1|
| ES001| French| 13| 2|
| ES003| French| 11| 3|
| ES003| Maths| 19| 1|
| ES002| Maths| 17| 2|
| ES001| Maths| 15| 3|
+-------+-------+-----+-------+
~~~~rowsBetween
This function helps us define dynamic windows that depend on the row in which a calculation
is being made and requires two elements that indicate the beginning and the end of the
dynamic window. These two values are integers relative to the current row, for example
rowsBetween(-1, 1) would indicate that the window goes from the row before the current row
to the row after the current row. However, Apache Spark provides and recommends using,
where possible, several functions to define these intervals, in order to minimise human error
and increase the readability of the code:
- currentRow indicates the current row where the calculation is being performed.
- unboundedPreceding indicates the first row of the group.
- unboundedFollowing indicates the last row of the group.
To observe how rowsBetween works, let’s assume we have the following data about the
price of a stock:
~~~~ python
prices = [('AAPL', '2021-01-01', 110),
('AAPL', '2021-01-02', 120),
('AAPL', '2021-01-03', 110),
('AAPL', '2021-01-04', 115),
('AAPL', '2021-01-05', 150),
('MSFT', '2021-01-01', 150),
('MSFT', '2021-01-02', 130),
('MSFT', '2021-01-03', 140),
('MSFT', '2021-01-04', 120),
('MSFT', '2021-01-05', 140)]
pricesSpark = spark.createDataFrame(prices, schema=['company', 'date', 'price'])\
.withColumn('date', to_date(col('date'))).cache()
pricesSpark.show()
~~~~
~~~~ console
+-------+----------+-----+
|company| date|price|
+-------+----------+-----+
| AAPL|2021-01-01| 110|
| AAPL|2021-01-02| 120|
| AAPL|2021-01-03| 110|
| AAPL|2021-01-04| 115|
| AAPL|2021-01-05| 150|
| MSFT|2021-01-01| 150|
| MSFT|2021-01-02| 130|
| MSFT|2021-01-03| 140|
| MSFT|2021-01-04| 120|
| MSFT|2021-01-05| 140|
+-------+----------+-----+
~~~~If we want to calculate a moving average from the first date to the date in the current row, we
could do the following in the current row:
~~~~ python
win = Window.partitionBy('company').orderBy('date').rowsBetween(Window.unboundedPreceding, Window.currentRow)
win2 = Window.partitionBy('company').orderBy('date')
pricesSpark.withColumn('histAvg', avg(col('price')).over(win))\
.withColumn('histAvg2', avg(col('price')).over(win2)).show()
~~~~
~~~~ console
+-------+----------+-----+------------------+------------------+
|company| date|price| histAvg| histAvg2|
+-------+----------+-----+------------------+------------------+
| AAPL|2021-01-01| 110| 110.0| 110.0|
| AAPL|2021-01-02| 120| 115.0| 115.0|
| AAPL|2021-01-03| 110|113.33333333333333|113.33333333333333|
| AAPL|2021-01-04| 115| 113.75| 113.75|
| AAPL|2021-01-05| 150| 121.0| 121.0|
| MSFT|2021-01-01| 150| 150.0| 150.0|
| MSFT|2021-01-02| 130| 140.0| 140.0|
| MSFT|2021-01-03| 140| 140.0| 140.0|
| MSFT|2021-01-04| 120| 135.0| 135.0|
| MSFT|2021-01-05| 140| 136.0| 136.0|
+-------+----------+-----+------------------+------------------+
~~~~See that using the win window or the win 2 window it gives the same result because in this
case there are no repeated dates within each group, so the behaviour of orderBy does not manifest itself. However, if we use the example with data from the PIB we notice that when
using rowsBetween the data used to calculate the sum is only up to the current row, so that
in contributionMostGreatest2 the telecommunication value is only 65 = 40 + 25 :
~~~~ python
win = Window().orderBy(col('percGdp').desc())
win2 = Window().orderBy(col('percGdp').desc()).rowsBetween(Window.unboundedPreceding, Window.currentRow)
gdpSpark.withColumn('percBiggerSectors', sum(col('percGdp')).over(win))\
.withColumn('percBiggerSectors2', sum(col('percGdp')).over(win2)).show()
~~~~
~~~~ console
+-------------+-------+-----------------+------------------+
| sector|percGdp|percBiggerSectors|percBiggerSectors2|
+-------------+-------+-----------------+------------------+
| Tourism| 40| 40| 40|
| Telecom| 25| 90| 65|
|Petrochemical| 25| 90| 90|
| Agriculture| 5| 100| 95|
| Construction| 5| 100| 100|
+-------------+-------+-----------------+------------------+
~~~~rangeBetween
This function is similar to rowsBetween but works with the values of the variable used inside
orderBy instead of the rows. A window created with rangeBetween has the same behaviour
as when using orderBy alone but it allows us to define a different range and not only from
the first value to the current one. It receives two arguments that indicate how many values
below and above the current value you want within the window.
For example, if the salary you have for a row is 10 euros and you have a window defined by
window.orderBy(‘salary’).rangeBetween(-5, 5); if we carry out a calculation on this window
the set that will be taken into account for this row will be all the rows whose wages are
between 5 and 15 euros.
Using the students’ data we might want to compare them with those students who get similar
grades, for example we could compare them with the average of their class but only those
students who get +-2 points more than them. Using rangeBetween the code would be:
~~~~ python
win = Window.partitionBy('course').orderBy('grade').rangeBetween(-2, 2)
result = gradesSpark.withColumn('similarStudentsAvg', mean(col('grade')).over(win))
result = result.withColumn('aboveAvg', when(col('grade')>=col('similarStudentsAvg'), lit(1)).otherwise(lit(0)))
result.show()
~~~~
~~~~ console
+-------+-------+-----+------------------+--------+
|student| course|grade|similarStudentsAvg|aboveAvg|
+-------+-------+-----+------------------+--------+
| ES002|English| 14| 14.0| 1|
| ES001|English| 18| 18.5| 0|
| ES003|English| 19| 18.5| 1|
| ES003| French| 11| 12.0| 0|
| ES001| French| 13| 12.0| 1|
| ES002| French| 16| 16.0| 1|
| ES001| Maths| 15| 16.0| 0|
| ES002| Maths| 17| 17.0| 1|
| ES003| Maths| 19| 18.0| 1|
+-------+-------+-----+------------------+--------+
~~~~Functions to create variables with windows
In Apache Spark we can divide the functions that can be used on a window into two main
groups. In addition, users can define their own functions, just like when using
groupBy (the use of udfs should be avoided as they tend to perform very poorly).
Analytical functions
They return a value for each row of a set, which may be different for each row.
- rank → Allows us to assign to each row within a group a natural number that indicates the
place the row has according to the value of the variable used in the orderBy. If two rows are
tied, they are assigned the same number in the rank and a number is skipped for the next
row. - denseRank → Has the same purpose as rank only that when there are ties a number is not
skipped for the row after the tied rows. - percentRank → Gives us the percentile that corresponds to the row when the data is
sorted according to the variable used in orderBy. - rowNumber → Allows us to assign to each row within a group a different natural number
that indicates the place that the row has according to the value of the variable used in the
orderBy. If two rows are tied the ranking is assigned randomly so the results are not always
equal. - nTile(n) → Used to distribute the rows of a group among n groups. It is very useful when
you want to make an ordinal variable from a numeric variable; for example 4 income levels
with the same number of observations each. - lag(col, n, default) → Gives us the value of the column col, n rows before the current row.
If n is a negative number it implies that we get the value that the column col, n rows after the
current row has. If the group does not have enough rows before or after the current row lag
returns nan unless the function is called with the optional parameter default in which case it
will return said value. The order of the rows is given by the column used in the window’s
orderBy so when using lag it is necessary that the window contains this clause. - lead(col, n, default) → Same as lag only that positive numbers imply later rows and
negative numbers earlier rows. - cumeDist → Gives the cumulative probability of each row according to the variable used in
orderBy. In other words, it tells us the proportion of rows whose value in the variable used in
orderBy is less than or equal to the value of that variable for the current row. - last(col, ignorenulls) → Returns the last value observed in the col column before the current row within the specified window. Requires the window to have an orderBy. The argument ignorenulls=True allows ignoring all missing values which causes the first non-null value before the current row to be returned.
Aggregation functions
All the functions that compact a set of data into a single value that represents the set, such
as sum, min, max, avg and count functions.
Additional usage examples
rangeBetween and window dates
To apply rangeBetween to create window dates, simply transform the dates to timestamp
and divide by the number of seconds in the period we want to calculate the window, for
example dividing by 86400 for days or 3600 for hours. In the following example, if we wanted
to find the three day moving average we couldn’t use rowsBetween as before because we
have missing dates but we can use rangeBetween:
~~~~ python
prices2 = [('AAPL', '2021-01-05', 110),
('AAPL', '2021-01-07', 120),
('AAPL', '2021-01-13', 110),
('AAPL', '2021-01-14', 115),
('AAPL', '2021-01-15', 150),
('MSFT', '2021-01-01', 150),
('MSFT', '2021-01-05', 130),
('MSFT', '2021-01-06', 140),
('MSFT', '2021-01-09', 120),
('MSFT', '2021-01-12', 140)]
pricesSpark2 = spark.createDataFrame(prices2, schema=['company', 'date', 'price'])\
.withColumn('date', to_date(col('date'))).cache()
pricesSpark2.show()
~~~~
~~~~ console
+-------+----------+-----+
|company| date|price|
+-------+----------+-----+
| AAPL|2021-01-05| 110|
| AAPL|2021-01-07| 120|
| AAPL|2021-01-13| 110|
| AAPL|2021-01-14| 115|
| AAPL|2021-01-15| 150|
| MSFT|2021-01-01| 150|
| MSFT|2021-01-05| 130|
| MSFT|2021-01-06| 140|
| MSFT|2021-01-09| 120|
| MSFT|2021-01-12| 140|
+-------+----------+-----+
~~~~
~~~~ python
days = lambda i: i * 86400
win = Window.partitionBy('company')\
.orderBy(col('date').cast("timestamp").cast("long"))\
.rangeBetween(-days(3), 0)
pricesSpark2.withColumn('movingAvg3d', avg(col('price')).over(win)).show()
~~~~
~~~~ console
+-------+----------+-----+-----------+
|company| date|price|movingAvg3d|
+-------+----------+-----+-----------+
| AAPL|2021-01-05| 110| 110.0|
| AAPL|2021-01-07| 120| 115.0|
| AAPL|2021-01-13| 110| 110.0|
| AAPL|2021-01-14| 115| 112.5|
| AAPL|2021-01-15| 150| 125.0|
| MSFT|2021-01-01| 150| 150.0|
| MSFT|2021-01-05| 130| 130.0|
| MSFT|2021-01-06| 140| 135.0|
| MSFT|2021-01-09| 120| 130.0|
| MSFT|2021-01-12| 140| 130.0|
+-------+----------+-----+-----------+
~~~~Creating lags for time series models
If we have a dataset that has no missing dates then obtaining a lagged variable is as simple
as using lag with the desired window. Let’s exemplify by applying this procedure to the set of
pricesSpark:
~~~~ python
days = lambda i: i * 86400
win = Window.partitionBy('company')\
.orderBy(col('date').cast("timestamp").cast("long"))
pricesSpark2.withColumn('price_1', lag(col('price')).over(win)).show()
~~~~
~~~~ console
+-------+----------+-----+-------+
|company| date|price|price_1|
+-------+----------+-----+-------+
| AAPL|2021-01-05| 110| null|
| AAPL|2021-01-07| 120| 110|
| AAPL|2021-01-13| 110| 120|
| AAPL|2021-01-14| 115| 110|
| AAPL|2021-01-15| 150| 115|
| MSFT|2021-01-01| 150| null|
| MSFT|2021-01-05| 130| 150|
| MSFT|2021-01-06| 140| 130|
| MSFT|2021-01-09| 120| 140|
| MSFT|2021-01-12| 140| 120|
+-------+----------+-----+-------+
~~~~However, in the case of the dataset pricesSpark2 it’s not possible to implement this
methodology since the value of the previous row can correspond to either 1 day before or
several days before. In this case we could use any aggregation function (e.g. sum) together
with a window containing rangeBetween to limit the dates on which the function is
calculated to exactly the number of previous days we want:
~~~~ python
days = lambda i: i * 86400
win = Window.partitionBy('company')\
.orderBy(col('date').cast("timestamp").cast("long"))\
.rangeBetween(-days(1),-days(1))
pricesSpark2.withColumn('price_1', sum(col('price')).over(win)).show()
~~~~
~~~~ console
+-------+----------+-----+-------+
|company| date|price|price_1|
+-------+----------+-----+-------+
| AAPL|2021-01-05| 110| null|
| AAPL|2021-01-07| 120| null|
| AAPL|2021-01-13| 110| null|
| AAPL|2021-01-14| 115| 110|
| AAPL|2021-01-15| 150| 115|
| MSFT|2021-01-01| 150| null|
| MSFT|2021-01-05| 130| null|
| MSFT|2021-01-06| 140| 130|
| MSFT|2021-01-09| 120| null|
| MSFT|2021-01-12| 140| null|
+-------+----------+-----+-------+
~~~~Time that has passed since an event
In many situations we find ourselves with datasets where in each row we have an event and
we want to know the date between some of them. For example, we could have a dataset
with the logs of certain machines along with the failures they have had:
~~~~ python
events = [('m1', '2021-01-01', 'e1'),
('m1', '2021-02-02', 'e1'),
('m1', '2021-03-03', 'e2'),
('m1', '2021-03-04', 'critical'),
('m1', '2021-07-05', 'e1'),
('m2', '2021-01-01', 'e1'),
('m2', '2021-06-02', 'critical'),
('m2', '2021-08-03', 'e2'),
('m2', '2021-09-04', 'e1'),
('m2', '2021-10-17', 'critical'),
('m2', '2021-10-25', 'e1')]
eventsSpark = spark.createDataFrame(events, schema=['machine', 'date', 'error'])\
.withColumn('date', to_date(col('date'))).cache()
eventsSpark.show()
~~~~
~~~~ console
+-------+----------+--------+
|machine| date| error|
+-------+----------+--------+
| m1|2021-01-01| e1|
| m1|2021-02-02| e1|
| m1|2021-03-03| e2|
| m1|2021-03-04|critical|
| m1|2021-07-05| e1|
| m2|2021-01-01| e1|
| m2|2021-06-02|critical|
| m2|2021-08-03| e2|
| m2|2021-09-04| e1|
| m2|2021-10-17|critical|
| m2|2021-10-25| e1|
+-------+----------+--------+
~~~~If we are interested in the time that has elapsed from each of the error types to a critical error
we could use the last function along with some auxiliary variables that have missing values
unless the error is of a certain type:
~~~~ python
errors = ['e1', 'e2']
# We create variables that have a value if the error is of a certain type
for i in errors:
eventsSpark = eventsSpark.withColumn('date_'+i, when(col('error')==i, col('date')).otherwise(None))
# We calculate the time from each type of error until a critical error
win = Window.partitionBy('machine').orderBy('date')
for i in errors:
eventsSpark = eventsSpark.withColumn('daysFromLast_'+i, last(col('date_'+i), True).over(win))\
.withColumn('daysFromLast_'+i,
(unix_timestamp(col('date'))-unix_timestamp(col('daysFromLast_'+i)))/86400)
eventsSpark.filter(col('error')=='critical').show()
~~~~
~~~~ console
+-------+----------+--------+-------+-------+-----------+-----------+-----------+-----------+---------------+---------------+
|machine| date| error|date_e1|date_e2|dateLast_e1|timeLast_e1|dateLast_e2|timeLast_e2|daysFromLast_e1|daysFromLast_e2|
+-------+----------+--------+-------+-------+-----------+-----------+-----------+-----------+---------------+---------------+
| m1|2021-03-04|critical| null| null| 2021-02-02| 30.0| 2021-03-03| 1.0| 30.0| 1.0|
| m2|2021-06-02|critical| null| null| 2021-01-01| 152.0| null| null| 152.0| null|
| m2|2021-10-17|critical| null| null| 2021-09-04| 43.0| 2021-08-03| 75.0| 43.0| 75.0|
+-------+----------+--------+-------+-------+-----------+-----------+-----------+-----------+---------------+---------------+
~~~~Conclusion
In this post we have learned what are windows in Spark, the syntax for their definition and the most common functions used with them. In addition, we have seen examples of tasks that can be solved with windows as well as the advantages of its use.
In future posts we will talk about how to define a simple demand model and a recursive strategy to find the optimal selling price. In the meantime, we encourage you to visit the Data Science category of the Damavis blog and see other articles similar to this one.
