Approach: use window functions to compute time gaps per user (difference between current event_time and previous event_time). Mark session boundaries where gap > 30 minutes, compute a cumulative session counter per user to create session_id, then aggregate per session. This is in-place, single-pass per partition and scales well with Spark when data is partitioned by user and sorted by event_time.python
from pyspark.sql import functions as F
from pyspark.sql.window import Window
# df: columns user_id (string), event_time (timestamp), bytes (int)
timeout = F.expr("INTERVAL 30 MINUTES")
# ensure partitioning and ordering
w = Window.partitionBy("user_id").orderBy("event_time")
# compute previous event_time and gap
df2 = (df
.withColumn("prev_ts", F.lag("event_time").over(w))
.withColumn("gap", F.when(F.col("prev_ts").isNull(), F.lit(0))
.otherwise(F.col("event_time").cast("long") - F.col("prev_ts").cast("long")))
.withColumn("new_session", F.when((F.col("prev_ts").isNull()) | (F.col("gap") > 30*60), 1).otherwise(0))
)
# cumulative session number per user
w2 = Window.partitionBy("user_id").orderBy("event_time").rowsBetween(Window.unboundedPreceding, 0)
df3 = df2.withColumn("session_num", F.sum("new_session").over(w2))
# build session_id and aggregate
sessions = (df3
.withColumn("session_id", F.concat_ws("_", F.col("user_id"), F.col("session_num").cast("string")))
.groupBy("session_id", "user_id", "session_num")
.agg(
F.min("event_time").alias("session_start"),
F.max("event_time").alias("session_end"),
F.count("*").alias("event_count"),
F.sum("bytes").alias("total_bytes")
)
.select("session_id", "user_id", "session_start", "session_end", "event_count", "total_bytes")
)
