Real-world PySpark skills

Harshal

3 min read

·

Jan 2, 2026

--

Listen

Share

Sample Input

Press enter or click to view image in full size

Expected output:

Press enter or click to view image in full size

Problem statement:

You work for a logistics platform processing delivery events.
Each order generates multiple events over time (CREATED, PICKED, DELIVERED, CANCELLED).

Identify pickup and delivery timestamps per order
Exclude cancelled and incomplete orders
Calculate delivery duration
Compute average delivery duration per city

Solution:

Sample data

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from datetime import datetime

spark = SparkSession.builder.getOrCreate()

# Define schema
schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("driver_id", StringType(), True),
    StructField("event_type", StringType(), True),
    StructField("event_ts", TimestampType(), True),
    StructField("city", StringType(), True)
])

# Sample data
data = [
    ("O1", "D1", "CREATED",   datetime(2024, 1, 1, 10, 0, 0), "Mumbai"),
    ("O1", "D1", "PICKED",    datetime(2024, 1, 1, 10, 15, 0), "Mumbai"),
    ("O1", "D1", "DELIVERED", datetime(2024, 1, 1, 11, 0, 0), "Mumbai"),

    ("O2", "D2", "CREATED",   datetime(2024, 1, 1, 9, 30, 0), "Delhi"),
    ("O2", "D2", "PICKED",    datetime(2024, 1, 1, 9, 50, 0), "Delhi"),
    ("O2", "D2", "CANCELLED", datetime(2024, 1, 1, 10, 5, 0), "Delhi"),

    ("O3", "D3", "CREATED",   datetime(2024, 1, 1, 8, 0, 0), "Mumbai"),
    ("O3", "D3", "PICKED",    datetime(2024, 1, 1, 8, 10, 0), "Mumbai"),
    ("O3", "D3", "DELIVERED", datetime(2024, 1, 1, 8, 40, 0), "Mumbai")
]

# Create DataFrame
df = spark.createDataFrame(data, schema)

df.show(truncate=False)

Get cancelled orders:

from pyspark.sql import functions as F

cancelled_orders = (
    df.filter(F.col("event_type") == "CANCELLED")
      .select("order_id")
      .distinct()
)

Filter our rows for cancelled orders:

df_valid = df.join(cancelled_orders, "order_id", "left_anti")

Press enter or click to view image in full size

From multiple rows per order
To one row per order with:

first pickup time and first delivery time

order_events = (
    df_valid
    .groupBy("order_id", "city")
    .agg(
        F.min(F.when(F.col("event_type") == "PICKED", F.col("event_ts")))
            .alias("pickup_time"),
        F.min(F.when(F.col("event_type") == "DELIVERED", F.col("event_ts")))
            .alias("delivery_time")
    )
)

Press enter or click to view image in full size

Filter data : keep only records with valid data.

complete_orders = order_events.filter(
    F.col("pickup_time").isNotNull() &
    F.col("delivery_time").isNotNull()
)

Press enter or click to view image in full size

Calculate Delivery Duration:

with_duration = complete_orders.withColumn(
    "delivery_duration_minutes",
    (F.col("delivery_time").cast("long") -
     F.col("pickup_time").cast("long")) / 60
)

Press enter or click to view image in full size

Aggregare based on city:

result = (
    with_duration
    .groupBy("city")
    .agg(
        F.round(F.avg("delivery_duration_minutes"), 2)
        .alias("avg_delivery_minutes")
    )
)

Press enter or click to view image in full size

Conclusion

This single question tests multiple real-world PySpark skills:

• Event-based data modeling
• Conditional aggregation (when + min)
• Anti-joins for exclusion
• Data quality handling
• Explaining transformations step by step

💡 Interview tip:
If you can explain what the DataFrame looks like after each transformation, you’re already ahead of most candidates.