metrics.py

import pyspark.sql.functions as F
from pyspark.sql import DataFrame
from config import Config
import logging

logger = logging.getLogger(__name__)

# Compute basic metrics using windowed aggregation
def compute_basic_metrics(logs_df: DataFrame) -> DataFrame:
    aggregated_df = logs_df.withWatermark("timestamp", Config.WATERMARK_DELAY) \
        .groupBy(
            F.window("timestamp", Config.WINDOW_DURATION, Config.SLIDE_DURATION),
            F.col("user_id")
        ).agg(
            F.count("activity").alias("activity_count"),
            F.sum("duration").alias("total_duration"),
            F.sum(F.col("error").cast("int")).alias("error_count")
        )
    logger.info("Basic metrics computed using window aggregation.")
    return aggregated_df

# Compute session metrics such as session duration and engagement rate
def compute_session_metrics(logs_df: DataFrame) -> DataFrame:
    session_metrics = logs_df.groupBy("user_id", "session_id").agg(
        F.min("timestamp").alias("session_start"),
        F.max("timestamp").alias("session_end"),
        F.sum("duration").alias("session_duration"),
        F.count("activity").alias("session_activity_count")
    )
    
    # Calculate engagement rate as activity count per duration unit (handle zero duration)
    session_metrics = session_metrics.withColumn(
        "engagement_rate",
        F.when(F.col("session_duration") > 0,
               F.col("session_activity_count") / F.col("session_duration"))
         .otherwise(0)
    )
    logger.info("Session metrics computed for each user session.")
    return session_metrics

# Combine basic and session metrics for a complete view
def compute_combined_metrics(logs_df: DataFrame) -> DataFrame:
    basic_metrics = compute_basic_metrics(logs_df)
    session_metrics = compute_session_metrics(logs_df)
    
    # Join based on user_id and approximate window overlap (this is a simple join example)
    combined_df = basic_metrics.join(session_metrics, on="user_id", how="left")
    logger.info("Combined metrics created by joining basic and session metrics.")
    return combined_df