نگهداشت (retention) بر اساس کوهورت ماهانه

from pyspark.sql import functions as F

base = commandes.withColumn("mois", F.trunc("order_date", "month"))
cohorte = base.groupBy("customer_id").agg(F.min("mois").alias("cohorte"))

retention = (
    base.join(cohorte, "customer_id")
        .withColumn("age_mois", F.months_between("mois", "cohorte").cast("int"))
        .groupBy("cohorte").pivot("age_mois", [0, 1, 2, 3])
        .agg(F.countDistinct("customer_id"))
)

for m in [1, 2, 3]:
    retention = retention.withColumn(
        f"M{m}", F.round(F.col(str(m)) / F.col("0") * 100, 1))

retention.select("cohorte", F.col("0").alias("clients"), "M1", "M2", "M3") \
         .orderBy("cohorte").show()