Objective

A streaming company wants to analyze its users' behavior alongside their subscription data to understand engagement. You have been provided with a UserBehavior log and a Subscription ledger.

Task

Calculate the total number of minutes each user watched only while they had an active subscription.

Assume a user's subscription is active on the exact day it starts and the exact day it ends. If a user watched content on a date when they didn't have an active subscription, do not include that watch time in the total. Handle the edge case where the subscriptionEnd is listed as "ongoing". Each user's subscription periods are guaranteed to be non-overlapping.

Save your resulting DataFrame as result_df. Ensure the output matches the exact schema order requested.

File Path

• Behavior Dataset: /home/interview/behavior.csv
• Subscription Dataset: /home/interview/subscription.csv
• Starter script: /home/interview/watch_time.py

Schema

behavior.csv

subscription.csv

Expected Output Schema

Example

Given this sample input:

UserBehavior

Subscription

The output would be:

Explanation: * U1 watched 45 mins on 01-01 and 60 mins on 01-02. Both fall within their active window (01-01 to 01-10), total = 105.

• U2 watched 70 mins on 01-03 and 50 mins on 01-02. Both fall within their active window (started 01-02 and is ongoing), total = 120.

• U3 watched 30 mins on 01-04. This falls within their active window (12-25 to 01-05), total = 30.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("PrepareshSpark").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

behavior = spark.read.csv("/home/interview/behavior.csv", header=True, inferSchema=True)
subscription = spark.read.csv("/home/interview/subscription.csv", header=True, inferSchema=True)

# Step 1: Join the DataFrames on userId
joined_df = behavior.join(subscription, on="userId", how="inner")

# Step 2: Filter for valid watch dates using conditional logic
valid_views = joined_df.filter(
    (F.col("date") >= F.col("subscriptionStart")) & 
    (
        (F.col("subscriptionEnd") == "ongoing") | 
        (F.col("date") <= F.col("subscriptionEnd"))
    )
)

# Step 3: Group by user and sum the valid watch durations
result_df = valid_views.groupBy("userId") \
    .agg(F.sum("watchDuration").cast("integer").alias("totalWatchTime")) \
    .select("userId", "totalWatchTime")

# --- Do not edit below this line ---
result_df.coalesce(1).write.csv("/home/interview/output", header=True, mode="overwrite")
spark.stop()

Explanation

Step 1: Joining the DataFrames

joined_df = behavior.join(subscription, on="userId", how="inner")

To determine if a specific watch event falls within a user's subscription window, we must first bring the two tables together. We perform an inner join on userId. This pairs every single viewing session with the corresponding subscription details for that user.

Step 2: Filtering by Date Ranges and "Ongoing" Status

valid_views = joined_df.filter(
    (F.col("date") >= F.col("subscriptionStart")) & 
    (
        (F.col("subscriptionEnd") == "ongoing") | 
        (F.col("date") <= F.col("subscriptionEnd"))
    )
)

This is the core logic of the problem. PySpark can natively compare string-formatted dates (like "2023-01-01") alphabetically, which conveniently works exactly like chronological date comparison.

• First, we ensure the view date is greater than or equal to subscriptionStart.
• Next, we must handle the upper bound. Because the subscriptionEnd column can contain the string "ongoing", simply checking date <= subscriptionEnd would crash or fail. We use an | (OR) operator to say: "Either the subscription is ongoing, OR the date is less than or equal to the end date."

Step 3: Aggregating the Valid Time

result_df = valid_views.groupBy("userId") \
    .agg(F.sum("watchDuration").cast("integer").alias("totalWatchTime"))

Once the invalid, out-of-bounds viewing sessions have been .filter()'d out, all that remains is to group the data by userId and sum the watchDuration. We chain .cast("integer") to ensure the final output strictly matches the integer requirement of the Output Schema.