如何在sparklyr中实现时间窗口内的唯一设备数统计（附data.table实现示例）

Got it, let's break down how to replicate your data.table rolling unique device count logic in sparklyr. First, let's recap what your original code does: for each adjusted timestamp (+540 seconds), you're counting unique devices that appeared in the window from timestamp - 540 seconds up to the original timestamp (effectively a 10-minute lookback window).

Step 1: Set Up Spark Connection & Load Data #

First, we'll get your sample data into Spark using sparklyr:

library(sparklyr)
library(dplyr)

# Initialize local Spark connection (adjust master for cluster use)
sc <- spark_connect(master = "local")

# Sample data matching your example
df <- data.frame(
  device_subscriber_id = c("x","a","z","x","a","z","x","y","a","z"),
  start_timestamp = as.POSIXct(
    c("2020-12-11 14:21:00","2020-12-11 14:22:00","2020-12-11 14:23:00",
      "2020-12-11 14:26:00","2020-12-11 14:24:00","2020-12-11 14:25:00",
      "2020-12-11 14:26:00","2020-12-11 14:28:00","2020-12-11 14:31:00","2020-12-11 14:38:00"),
    format = "%Y-%m-%d %H:%M:%S"
  )
)

# Copy data to Spark cluster
spark_df <- copy_to(sc, df, name = "device_data", overwrite = TRUE)

Method 1: Range Join (Mirroring data.table Logic) #

This approach directly mirrors your data.table cross-join with range condition. We'll generate window ranges for each timestamp, then join to count unique devices in each window:

# 1. Create a table of window ranges and target timestamps (matches your data.table's derived columns)
window_tbl <- spark_df %>%
  mutate(
    window_start = start_timestamp - expr("INTERVAL 540 SECOND"),  # start_timestamp - 540s
    window_end = start_timestamp,
    target_time = start_timestamp + expr("INTERVAL 540 SECOND")   # Final timestamp for grouping
  ) %>%
  select(window_start, window_end, target_time)

# 2. Perform range join (equivalent to data.table's `on = .(start_timestamp >= start_timestamp2, start_timestamp<=start_timestamp3)`)
joined_tbl <- spark_df %>%
  inner_join(
    window_tbl,
    join_by(start_timestamp >= window_start, start_timestamp <= window_end)
  )

# 3. Calculate unique devices per target timestamp
result_tbl <- joined_tbl %>%
  group_by(target_time) %>%
  summarise(unique_devices_seen = n_distinct(device_subscriber_id)) %>%
  rename(start_timestamp = target_time) %>%
  arrange(start_timestamp)

# View results
collect(result_tbl)

Method 2: Sliding Window Functions (More Efficient for Big Data) #

For large datasets, Spark's built-in sliding window functions are more performant than range joins. We'll use collect_set to gather unique devices in the window, then count them:

# Define a sliding window: range from current timestamp -540s to current timestamp
window_spec <- window(
  order_by = start_timestamp,
  rangeBetween(-540, 0)  # Units are seconds for timestamp columns
)

# Calculate unique devices in each window and map to your target timestamp
result_window_tbl <- spark_df %>%
  mutate(
    # Collect unique devices in the sliding window
    unique_devices = collect_set(device_subscriber_id) over window_spec,
    # Compute target timestamp (matches your data.table's `start_timestamp + 540`)
    target_time = start_timestamp + expr("INTERVAL 540 SECOND"),
    # Count unique devices
    unique_devices_seen = size(unique_devices)
  ) %>%
  # Deduplicate since multiple rows may map to the same target timestamp
  distinct(target_time, unique_devices_seen) %>%
  rename(start_timestamp = target_time) %>%
  arrange(start_timestamp)

# View results
collect(result_window_tbl)

Both methods will return results matching your expected_dt output. The sliding window method is preferred for large-scale data as it avoids expensive cross-joins.

Clean Up #

Don't forget to disconnect from Spark when done:

spark_disconnect(sc)

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