Cohort Analysis and Retention Metrics Questions

from pyspark.sql import functions as F, Window

# 1) load tables
users = spark.table("users")  # contains user_id, first_touch_date
events = spark.table("events")  # large clickstream: user_id, event_ts

# 2) pre-aggregate events to user-day to dedupe
events_day = (events
  .withColumn("event_date", F.to_date("event_ts"))
  .groupBy("user_id", "event_date")
  .agg(F.count("*").alias("events_count")))

# 3) join to acquisitions (only users with acquisition date)
joined = (users.select("user_id", F.to_date("first_touch_date").alias("acq_date"))
  .join(events_day, on="user_id", how="left")
  .withColumn("days_after", F.datediff("event_date", "acq_date"))
  .filter((F.col("days_after") >= 0) & (F.col("days_after") < 90)))

# 4) mark active (1 per user-day)
user_day_active = joined.select("user_id", "acq_date", "event_date", "days_after") \
  .withColumn("active", F.lit(1))

# 5) per acquisition_date and cohort_day compute unique users active
retention = (user_day_active
  .groupBy("acq_date", "days_after")
  .agg(F.countDistinct("user_id").alias("active_users")))

# 6) get cohort sizes and assemble retention table
cohort_sizes = users.groupBy(F.to_date("first_touch_date").alias("acq_date")) \
  .agg(F.countDistinct("user_id").alias("cohort_size"))

retention_table = (retention.join(cohort_sizes, on="acq_date")
  .withColumn("retention_rate", F.col("active_users") / F.col("cohort_size"))
  .orderBy("acq_date", "days_after"))

select cohort_date,
       sum(case when amount>0 then amount else 0 end) as gross_revenue,
       sum(amount) as net_revenue,
       sum(case when amount<0 and datediff(day, purchase_ts, event_ts)=0 then amount else 0 end) as same_day_refunds
from ledger
join users u on ledger.user_id = u.id
group by cohort_date;