Advanced window functions in PySpark for time-series analysis

Instruction: Explain how you would use PySpark's window functions to analyze time-series data, providing examples of complex queries.

Context: Candidates must showcase their deep understanding of window functions in PySpark, particularly in the context of time-series data analysis, including handling time windows, aggregations, and partitions.

Updated

A Window specification defines logical groups, ordering and a frame. rowsBetween counts rows; bounded rangeBetween offsets the value of a single numeric ordering expression. Ordering by a DateType column and passing -6 does not directly define a valid seven-day numeric range.

For one validated row per user/day, convert the date to an integer day index. With an existing session named spark:

from pyspark.sql import Window, functions as F

daily = spark.createDataFrame([
    ('u', '2026-01-01', 2.0), ('u', '2026-01-03', 6.0),
    ('u', '2026-01-08', 10.0), ('v', '2026-01-08', 20.0),
], 'user_id string, date_text string, views double')
daily = daily.withColumn('day', F.to_date('date_text')).withColumn(
    'day_index', F.datediff('day', F.lit('1970-01-01')))
frame = Window.partitionBy('user_id').orderBy('day_index').rangeBetween(-6, 0)
rolling = daily.withColumn('observed_day_mean', F.avg('views').over(frame))
assert [(r.user_id, r.date_text, r.observed_day_mean)
        for r in rolling.orderBy('user_id', 'day').collect()] == [
    ('u', '2026-01-01', 2.0), ('u', '2026-01-03', 4.0),
    ('u', '2026-01-08', 8.0), ('v', '2026-01-08', 20.0)
]

This averages observed days within the current date and previous six calendar dates. It does not fill absent days with zero or emit missing dates. Join a complete calendar first only if the data contract justifies that denominator. Validate unique user/day keys, dates and finite counts; duplicate peers change the weighting.

Signup-relative analysis needs a fixed anchor, not windows relative to every activity row. For example:

users = spark.createDataFrame([
    ('u', '2026-01-01'), ('v', '2026-01-01'), ('w', '2026-01-01')
], 'user_id string, signup_text string').withColumn('signup', F.to_date('signup_text'))
activity = spark.createDataFrame([
    ('u', '2026-01-01'), ('u', '2026-01-08'), ('v', '2026-01-02')
], 'user_id string, event_text string').withColumn('event_day', F.to_date('event_text'))
joined = users.join(activity, 'user_id', 'left').withColumn(
    'age_days', F.datediff('event_day', 'signup'))
flags = joined.groupBy('user_id').agg(
    F.max(F.when(F.col('age_days').between(0, 6), 1).otherwise(0)).alias('week1'),
    F.max(F.when(F.col('age_days').between(7, 13), 1).otherwise(0)).alias('week2'),
)
counts = flags.agg(F.sum('week1').alias('eligible'),
    F.sum(F.col('week1') * F.col('week2')).alias('retained')).first()
rate = counts.retained / counts.eligible if counts.eligible else None
assert rate == 0.5

Here retention explicitly means week-two activity among week-one-active users. Include only fully observable signup cohorts; use a consistent business time zone when deriving dates, and handle invalid IDs/timestamps before joining. Dividing week-two activity volume by week-one volume is an activity ratio, not this user-retention measure. For lag or row-number queries, provide a deterministic tie-breaker when timestamps tie.

Window frames, Date difference

Related Questions