Window Functions ​

WindowSpec ​

A window function, often referred to as a "window specification" or "windowing function," is a concept used in the context of database query languages like SQL and data analytics tools. It enables users to perform calculations on a specific "window" of data within a result set, rather than on the entire dataset. The window is defined by a set of rows related to the current row being processed.

The concept of windows and window functions is powerful for performing complex calculations and aggregations that involve data from multiple rows, without needing to perform explicit self-joins or subqueries. Window functions provide a more efficient and concise way to achieve these calculations.

The basic components of a window specification include:

1. Partition By Clause: This clause is used to divide the result set into partitions or groups based on specified columns. Each partition represents a separate window. The window function will be applied independently to each partition.

2. Order By Clause: This clause is used to define the ordering of rows within each partition. The window function will operate on the rows within each partition based on this order.

3. Window Frame Specification: The window frame defines the set of rows included in the window relative to the current row being processed. It is specified using "ROWS BETWEEN" or "RANGE BETWEEN" clauses. The default frame is usually "UNBOUNDED PRECEDING" to "CURRENT ROW," which means the window includes all rows from the beginning of the partition to the current row.

Common window functions include:

• Aggregate Functions: Functions like SUM, AVG, MIN, MAX, etc., can be used to calculate aggregates over the rows within the window.
• Ranking Functions: Functions like ROW_NUMBER, RANK, DENSE_RANK, and NTILE can assign ranks or row numbers to the rows within the window based on the specified ordering.
• Analytic Functions: Functions like LEAD and LAG allow accessing data from the next or previous rows within the window.
• Cumulative Functions: Functions like CUME_DIST, PERCENT_RANK, and PERCENTILE_CONT calculate cumulative distributions and percentiles.

An example of a window function is finding the average salary of employees within each department. Using a window function, we can partition the data by department, order it by salary, and calculate the average salary for each employee relative to their department.

The use of window functions can significantly simplify and optimize complex queries, making them a valuable tool in SQL-based data analysis and reporting.

Example ​

Suppose we have a DataFrame containing information about employees, including their names, departments, and salaries. We want to calculate the average salary for each department using a window function.

from pyspark.sql import SparkSession

# Create a SparkSession
spark = SparkSession.builder
.appName("Window Function Example")
.getOrCreate()

Next, let's create a sample DataFrame with some data:

from pyspark.sql import Row

# Sample data
data = [
    Row(name="Alice", department="HR", salary=40000),
    Row(name="Bob", department="HR", salary=50000),
    Row(name="Charlie", department="Finance", salary=60000),
    Row(name="David", department="Finance", salary=55000),
    Row(name="Eva", department="HR", salary=45000),
    Row(name="Frank", department="IT", salary=70000),
]

# Create DataFrame from the sample data
df = spark.createDataFrame(data)
df.show()

The output will be:

+-------+---------+-------+
|   name|department|salary|
+-------+----------+------+
|  Alice|       HR | 40000|
|    Bob|       HR | 50000|
|Charlie|  Finance | 60000|
|  David|  Finance | 55000|
|    Eva|       HR | 45000|
|  Frank|       IT | 70000|
+-------+----------+------+

Now, we can use a window function to calculate the average salary for each department:

from pyspark.sql.window import Window
from pyspark.sql import functions as F

# Define the window specification partitioned by 'department'
window_spec = Window.partitionBy("department")

# Calculate the average salary using the window function
df_with_avg_salary = df.withColumn("average_salary", F.avg("salary").over(window_spec))

df_with_avg_salary.show()

The output will be:

+-------+---------+------+--------------+
|   name|department|salary|average_salary|
+-------+---------+------+--------------+
|  Alice|       HR| 40000|       45000.0|
|    Bob|       HR| 50000|       45000.0|
|    Eva|       HR| 45000|       45000.0|
|Charlie|  Finance| 60000|       57500.0|
|  David|  Finance| 55000|       57500.0|
|  Frank|       IT| 70000|       70000.0|
+-------+---------+------+--------------+

As you can see, the average_salary column is calculated using the window function avg("salary").over(window_spec), which computes the average salary for each department separately. The window function partitions the data by the department column, and within each partition, it calculates the average salary.

This is a simple example of using window functions in PySpark. Window functions can be very powerful for complex analytical tasks that require aggregations and comparisons across multiple rows. They offer a flexible way to perform calculations on specific subsets of data within a DataFrame. Sure! Let's explain the other window functions in PySpark, categorized into the four sections as requested.

WindowSpec Functions ​

There a few more functions apart from partitionBy that can be used to define a windowSpec. See the official doc for more.

Example

Here's an example with rowsBetween

from pyspark.sql import SparkSession

# Create a SparkSession
spark = SparkSession.builder
    .appName("ROWS BETWEEN Clause Example")
    .getOrCreate()

Next, let's create a sample DataFrame with some data:

from pyspark.sql import Row

# Sample data
data = [
    Row(date="2023-07-01", revenue=1000),
    Row(date="2023-07-02", revenue=1500),
    Row(date="2023-07-03", revenue=800),
    Row(date="2023-07-04", revenue=1200),
    Row(date="2023-07-05", revenue=2000),
]

# Create DataFrame from the sample data
df = spark.createDataFrame(data)
df.show()

The output will be:

+----------+-------+
|      date|revenue|
+----------+-------+
|2023-07-01|   1000|
|2023-07-02|   1500|
|2023-07-03|    800|
|2023-07-04|   1200|
|2023-07-05|   2000|
+----------+-------+

Now, we can use the rowsBetween clause to calculate the cumulative revenue and the average revenue of the previous three rows for each row:

from pyspark.sql.window import Window
from pyspark.sql import functions as F

# Define the window specification ordered by 'date'
window_spec = Window.orderBy("date")

# Calculate the cumulative revenue using the 'ROWS BETWEEN' clause
df_with_cumulative_revenue = df.withColumn("cumulative_revenue",
                                           F.sum("revenue").over(window_spec.rowsBetween(Window.unboundedPreceding, 0)))

# Calculate the average revenue of the previous three rows using the 'ROWS BETWEEN' clause
df_with_average_revenue = df.withColumn("average_revenue", F.avg("revenue").over(window_spec.rowsBetween(-3, -1)))

df_with_cumulative_revenue.show()
df_with_average_revenue.show()

The output will be:

+----------+-------+------------------+
|      date|revenue|cumulative_revenue|
+----------+-------+------------------+
|2023-07-01|   1000|              1000|
|2023-07-02|   1500|              2500|
|2023-07-03|    800|              3300|
|2023-07-04|   1200|              4500|
|2023-07-05|   2000|              6500|
+----------+-------+------------------+

+----------+-------+---------------+
|      date|revenue|average_revenue|
+----------+-------+---------------+
|2023-07-01|   1000|           null|
|2023-07-02|   1500|         1000.0|
|2023-07-03|    800|         1250.0|
|2023-07-04|   1200|         1100.0|
|2023-07-05|   2000|         1166.7|
+----------+-------+---------------+

As you can see, we used the ROWS BETWEEN clause with appropriate offsets to calculate the cumulative revenue and the average revenue of the previous three rows for each row. The clause allows us to define the range of rows to be included in the window, which is useful for performing various advanced calculations based on specific sets of rows within the DataFrame.

📖👉 Official Doc

Aggregate Functions: ​

a. SUM(): ​

SUM() is an aggregate window function that calculates the sum of a numerical column within the window partition. It adds up all the values of the specified column for each row in the window partition.

Example: Suppose we want to calculate the total salary spent within each department.

from pyspark.sql.window import Window
from pyspark.sql import functions as F

window_spec = Window.partitionBy("department")

df_with_total_salary = df.withColumn("total_salary", F.sum("salary").over(window_spec))
df_with_total_salary.show()

b. AVG(): ​

AVG() is an aggregate window function that computes the average of a numerical column within the window partition. It calculates the average value of the specified column for each row in the window partition.

Example: Suppose we want to find the average salary of employees within each department.

window_spec = Window.partitionBy("department")

df_with_avg_salary = df.withColumn("avg_salary", F.avg("salary").over(window_spec))
df_with_avg_salary.show()

c. MIN() and MAX(): ​

MIN() and MAX() are aggregate window functions that find the minimum and maximum values of a column within the window partition, respectively.

Example: Suppose we want to find the minimum and maximum salary within each department.

window_spec = Window.partitionBy("department")

df_with_min_salary = df.withColumn("min_salary", F.min("salary").over(window_spec))
df_with_max_salary = df.withColumn("max_salary", F.max("salary").over(window_spec))

df_with_min_salary.show()
df_with_max_salary.show()

Ranking Functions: ​

In the previous response, we already covered ROW_NUMBER(), RANK(), and DENSE_RANK() as ranking functions.

Analytic Functions: ​

a. LEAD() and LAG(): ​

As explained earlier, LEAD() and LAG() are analytic functions that allow you to access data from subsequent or previous rows within the window partition. They are commonly used for time-based analysis and data comparisons.

Cumulative Functions: ​

a. CUME_DIST(): ​

CUME_DIST() is a cumulative distribution function that returns the cumulative distribution of a value within the window partition. It represents the relative position of a value compared to the entire partition.

Example: Suppose we want to calculate the cumulative distribution of salaries within each department.

window_spec = Window.partitionBy("department").orderBy("salary")

df_with_cume_dist = df.withColumn("cume_dist", F.cume_dist().over(window_spec))
df_with_cume_dist.show()

b. PERCENT_RANK(): ​

PERCENT_RANK() is a cumulative function that returns the relative rank of a value within the window partition. It represents the relative position of a value compared to the other values in the partition.

Example: Suppose we want to calculate the percent rank of salaries within each department.

window_spec = Window.partitionBy("department").orderBy("salary")

df_with_percent_rank = df.withColumn("percent_rank", F.percent_rank().over(window_spec))
df_with_percent_rank.show()

c. PERCENTILE_CONT() and PERCENTILE_DISC(): ​

PERCENTILE_CONT() and PERCENTILE_DISC() are cumulative window functions that compute the percentile value of a column within the window partition. PERCENTILE_CONT() returns a continuous value, while PERCENTILE_DISC() returns a discrete value.

Example: Suppose we want to find the 50th percentile (median) of salaries within each department.

window_spec = Window.partitionBy("department")

df_with_median = df.withColumn("median_salary", F.percentile_cont(0.5).over(window_spec))
df_with_median.show()