Spark SQL window function with complex condition

by

Spark >= 3.2

Recent Spark releases provide native support for session windows in both batch and structured streaming queries (see SPARK-10816 and its sub-tasks, especially SPARK-34893).

The official documentation provides nice usage example.

Spark < 3.2

Here is the trick. Import a bunch of functions:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{coalesce, datediff, lag, lit, min, sum}

Define windows:

val userWindow = Window.partitionBy("user_name").orderBy("login_date")
val userSessionWindow = Window.partitionBy("user_name", "session")

Find the points where new sessions starts:

val newSession =  (coalesce(
  datediff($"login_date", lag($"login_date", 1).over(userWindow)),
  lit(0)
) > 5).cast("bigint")

val sessionized = df.withColumn("session", sum(newSession).over(userWindow))

Find the earliest date per session:

val result = sessionized
  .withColumn("became_active", min($"login_date").over(userSessionWindow))
  .drop("session")

With dataset defined as:

val df = Seq(
  ("SirChillingtonIV", "2012-01-04"), ("Booooooo99900098", "2012-01-04"),
  ("Booooooo99900098", "2012-01-06"), ("OprahWinfreyJr", "2012-01-10"), 
  ("SirChillingtonIV", "2012-01-11"), ("SirChillingtonIV", "2012-01-14"),
  ("SirChillingtonIV", "2012-08-11")
).toDF("user_name", "login_date")

The result is:

+----------------+----------+-------------+
|       user_name|login_date|became_active|
+----------------+----------+-------------+
|  OprahWinfreyJr|2012-01-10|   2012-01-10|
|SirChillingtonIV|2012-01-04|   2012-01-04| <- The first session for user
|SirChillingtonIV|2012-01-11|   2012-01-11| <- The second session for user
|SirChillingtonIV|2012-01-14|   2012-01-11| 
|SirChillingtonIV|2012-08-11|   2012-08-11| <- The third session for user
|Booooooo99900098|2012-01-04|   2012-01-04|
|Booooooo99900098|2012-01-06|   2012-01-04|
+----------------+----------+-------------+

More Related Contents:

  1. Spark Window Functions – rangeBetween dates
  2. How to aggregate over rolling time window with groups in Spark
  3. How to select the first row of each group?
  4. Querying Spark SQL DataFrame with complex types
  5. Concatenate columns in Apache Spark DataFrame
  6. Avoid performance impact of a single partition mode in Spark window functions
  7. Does SparkSQL support subquery?
  8. Including null values in an Apache Spark Join
  9. Spark sql queries vs dataframe functions
  10. Explode (transpose?) multiple columns in Spark SQL table
  11. How to get other columns when using Spark DataFrame groupby?
  12. get TopN of all groups after group by using Spark DataFrame
  13. Why using a UDF in a SQL query leads to cartesian product?
  14. pyspark: count distinct over a window
  15. What is the difference between cube, rollup and groupBy operators?
  16. Difference between === null and isNull in Spark DataDrame
  17. PostgreSQL: running count of rows for a query ‘by minute’
  18. How to split Vector into columns – using PySpark
  19. Split Spark Dataframe string column into multiple columns
  20. How to change a dataframe column from String type to Double type in PySpark?
  21. pyspark dataframe filter or include based on list
  22. How to use a Scala class inside Pyspark
  23. Filter Pyspark dataframe column with None value
  24. Fill in null with previously known good value with pyspark
  25. Count number of non-NaN entries in each column of Spark dataframe in PySpark
  26. Load CSV file with PySpark
  27. What is the best way to remove accents with Apache Spark dataframes in PySpark?
  28. Multiple condition filter on dataframe
  29. How to connect HBase and Spark using Python?
  30. Apache Spark — Assign the result of UDF to multiple dataframe columns

Leave a Comment