42. Capstone 2 — customer analytics

A real analytics question: who is the top-spending customer in each segment? This needs a join (orders → customers), an aggregation (spend per customer), and a window (rank within segment) — Modules 2, 4, and 5 working together.

The brief

Return one row per segment (premium, standard) with the highest-spending customer in it:

1. Join deduplicated orders to customers on customer_id.
2. Compute each order's line_total = quantity * unit_price.
3. Total spend per customer (customer_id, name, segment), rounded.
4. Rank customers within each segment by spend, highest first.
5. Keep the top 1 per segment; return segment, name, spend.

Building it in stages

Stage 1 — spend per customer:

spend = (
    orders.dropDuplicates()
    .join(customers, "customer_id")
    .withColumn("line_total", F.col("quantity") * F.col("unit_price"))
    .groupBy("customer_id", "name", "segment")
    .agg(F.round(F.sum("line_total"), 2).alias("spend"))
)
spend.orderBy(F.col("spend").desc()).show(5)

Stage 2 — rank within segment and keep the winner. This is the "top-N per group" pattern from lesson 5.4, with N = 1.

Your turn

Finish the analysis: from the spend DataFrame idea above, rank by spend within each segment and keep rank 1. Return segment, name, spend. Assign to result.

w = Window.partitionBy("segment").orderBy(F.col("spend").desc())
spend = (
    orders.dropDuplicates()
    .join(customers, "customer_id")
    .withColumn("line_total", F.col("quantity") * F.col("unit_price"))
    .groupBy("customer_id", "name", "segment")
    .agg(F.round(F.sum("line_total"), 2).alias("spend"))
)
result = spend  # <- row_number().over(w) AS rn, keep rn == 1, select segment, name, spend
result.show()

w = Window.partitionBy("segment").orderBy(F.col("spend").desc())
spend = (
    orders.dropDuplicates()
    .join(customers, "customer_id")
    .withColumn("line_total", F.col("quantity") * F.col("unit_price"))
    .groupBy("customer_id", "name", "segment")
    .agg(F.round(F.sum("line_total"), 2).alias("spend"))
)
result = (
    spend.withColumn("rn", F.row_number().over(w))
    .filter(F.col("rn") == 1)
    .select("segment", "name", "spend")
)