Spark窗口函数多列应用：统计用户发送时间前2小时打开次数

Let's break down why your current code isn't producing the expected results and walk through the correct implementation step by step.

What's Wrong with Your Current Approach? #

Your code uses a rangeBetween window based on sendUnix, counting openUnix values in the window. Here's the critical issue:

• The window includes rows where sendTime falls within 2 hours before the current row's sendTime, but this doesn't account for whether the openTime itself is before the current row's sendTime. For example, the row with sendTime=15:00:00 has openTime=15:50:00—this open event happens after the send time, so it shouldn't be counted in the "pre-send open count".
• This leads to incorrect counts (like the first row showing 1 instead of 0) and overcounts for later rows.

Solution 1: Efficient Window Function Approach #

This method avoids expensive cross joins by using window functions to collect valid open times per user, then filtering those times to match the 2-hour pre-send window. It's ideal for large datasets.

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.TimestampType
import org.apache.spark.sql.Window

// Original DataFrame
val df = Seq(
  ("user1", "2018-04-05 15:00:00", "2018-04-05 15:50:00"),
  ("user1", "2018-04-05 16:00:00", "2018-04-05 16:50:00"),
  ("user1", "2018-04-05 17:00:00", "2018-04-05 17:50:00"),
  ("user1", "2018-04-05 18:00:00", "2018-04-05 18:50:00"),
  ("user2", "2018-04-05 15:00:00", "2018-04-05 15:50:00"),
  ("user2", "2018-04-05 16:00:00", "2018-04-05 16:50:00"),
  ("user2", "2018-04-05 17:00:00", "2018-04-05 17:50:00"),
  ("user2", "2018-04-05 17:30:00", "2018-04-05 17:40:00"),
  ("user2", "2018-04-05 18:00:00", null),
  ("user2", "2018-04-05 19:00:00", null)
).toDF("id", "sendTime", "openTime")

// Convert string timestamps to TimestampType for easy date arithmetic
val dfWithTs = df
  .withColumn("sendTs", col("sendTime").cast(TimestampType))
  .withColumn("openTs", col("openTime").cast(TimestampType))

// Collect all non-null open times per user into an array
val collectWindow = Window.partitionBy("id")
val dfWithOpenList = dfWithTs
  .withColumn("allOpenTimes", collect_list(when(col("openTs").isNotNull, col("openTs"))).over(collectWindow))

// Calculate count of open times that fall within 2 hours before the send time AND are before the send time
val result = dfWithOpenList
  .withColumn("count", size(
    filter(
      col("allOpenTimes"),
      openTs => openTs >= date_add(col("sendTs"), -2.0/24) && openTs < col("sendTs")
    )
  ))
  .select("id", "sendTime", "openTime", "count")

// View the final result
result.show()

Solution 2: Cross Join and Filter (For Smaller Datasets) #

If you're working with a smaller dataset, this approach is more straightforward: join the original DataFrame with valid open records, filter for the time range, and aggregate counts.

// Extract valid open records (exclude null openTime values)
val validOpens = dfWithTs
  .filter(col("openTs").isNotNull)
  .select("id", "openTs")
  .withColumnRenamed("id", "openId")

// Join with original DataFrame, filter for the 2-hour pre-send window, then count matching opens
val countDf = dfWithTs
  .join(validOpens, col("id") === col("openId"), "left")
  .filter(col("openTs") >= date_add(col("sendTs"), -2.0/24) && col("openTs") < col("sendTs"))
  .groupBy("id", "sendTime", "openTime")
  .agg(count("openTs").as("count"))

// Left join back to original DataFrame to fill in 0 counts for rows with no matching opens
val finalResult = dfWithTs
  .select("id", "sendTime", "openTime")
  .join(countDf, Seq("id", "sendTime", "openTime"), "left")
  .na.fill(0, Seq("count"))

finalResult.show()

Both solutions will generate your expected output. Solution 1 is more performant for large datasets, while Solution 2 is easier to read for smaller workloads.

内容的提问来源于stack exchange，提问作者Pooya