Measuring Cart Abandonment with Joins

In the previous part, we built daily product funnels. Those funnels gave us a broad view of how many sessions viewed a product, added it to the cart, started checkout, and completed a purchase.

Now let’s ask a more specific question.

When a product was added to a cart, was that same product purchased during the same session?

If it was not, we will treat that session-product pair as an abandoned cart. To find those cases, we will introduce one of the most important operations in data processing: the join.

Just like in the previous parts, we will implement the job twice:

• once in Spark SQL
• once in the Spark DataFrame API

Both versions will produce the same daily product-level result.

What Is Cart Abandonment?

In a retail website, users often add products to their cart without purchasing them.

They may change their mind. They may want to save a product for later. The price may feel too high. Shipping costs may be surprising. Or perhaps they were distracted by something very important, such as another browser tab.

Whatever the reason, we call this behavior cart abandonment.

For this tutorial, we will define cart abandonment at the session-product level:

A product is abandoned when it was added to the cart but was not purchased in the same session.

This is intentionally a simple definition.

If a product was added three times but purchased once, we will treat it as purchased rather than abandoned. We are only checking whether any purchase happened for that product in the session. Comparing quantities added and purchased would let us study partial abandonment, but that is a different and more detailed problem.

We will also ignore the exact ordering of the events here. If the session-product summary says both added_to_cart and purchased are true, we consider the cart completed.

Why Use a Join?

The retail_session_products table from Part 8 already has both of these Boolean columns on the same row:

• added_to_cart
• purchased

That means we could find abandoned carts with a direct filter:

WHERE added_to_cart AND NOT purchased

For this prepared table, that is the simpler solution.

So why are we using a join?

Because in many real systems, cart activity and purchase activity do not arrive in one convenient table. Cart events may come from the website, while purchase records may come from an order system. To determine whether a cart became a purchase, we need to connect related rows from those datasets.

In this tutorial, we will model that situation by splitting retail_session_products into two logical datasets:

• sessions that added a product to the cart
• sessions that purchased a product

Then we will join them back together.

This is slightly more work for our current table, but it gives us a useful pattern that applies to much more realistic data systems.

Understanding Inner and Left Joins

A join combines related rows from two datasets using one or more matching columns.

Let’s start with a small example.

Suppose our cart dataset contains these rows:

And our purchase dataset contains:

The cart for product-b in session-1 does not have a matching purchase.

Inner Join

An inner join keeps rows only when the join keys match on both sides.

If we inner-join the cart dataset with the purchase dataset, the result contains:

This tells us which carted products were purchased.

But it removes the unmatched session-1 and product-b row. That is exactly the row we need in order to measure abandonment.

Left Join

A left join keeps every row from the left dataset, even when there is no match on the right.

If the cart dataset is on the left, the joined result looks like this:

The unmatched cart row stays in the result. Columns supplied by the purchase side are null because Spark could not find a matching purchase.

That gives us the basic rule for this job:

• a matched purchase means the cart was completed
• a missing purchase match means the cart was abandoned

Why We Need Two Join Keys

Notice that session-1 contains two different products.

If we joined only on session_id, the purchase of product-a could incorrectly match the cart row for product-b. Spark would know that something was purchased in the session, but not that it was the same product.

That is why we join using both:

• session_id
• product_id

Together, those columns identify the product behavior we want to connect inside a session.

What We Are Going to Build

We will create a daily product-level cart-abandonment table.

Each row will contain:

• the number of sessions that added the product to a cart
• the number of those cart sessions that purchased the product
• the number of those cart sessions that abandoned the product
• the cart-abandonment rate

For every output row, this relationship should hold:

purchased cart sessions + abandoned cart sessions = total cart sessions

The abandonment rate is:

abandoned cart sessions / total cart sessions

As before, we will create separate destination tables for the SQL and DataFrame jobs so that their results do not get mixed together.

Creating the Destination Tables

Create the SQL destination table from the interactive shell:

spark.sql("""
    CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_daily_product_cart_abandonment (
      event_date DATE,

      product_id STRING,
      product_name STRING,
      category STRING,

      cart_session_count BIGINT,
      purchased_cart_session_count BIGINT,
      abandoned_cart_session_count BIGINT,

      cart_abandonment_rate DECIMAL(9, 4)
    )
    USING iceberg
""")

Now create the equivalent table for the DataFrame job:

spark.sql("""
    CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_daily_product_cart_abandonment_df (
      event_date DATE,

      product_id STRING,
      product_name STRING,
      category STRING,

      cart_session_count BIGINT,
      purchased_cart_session_count BIGINT,
      abandoned_cart_session_count BIGINT,

      cart_abandonment_rate DECIMAL(9, 4)
    )
    USING iceberg
""")

The rate is stored as a decimal value. For example, 0.7500 means that 75 percent of the product’s cart sessions were abandoned.

The jobs check that the cart count is greater than zero before dividing. In practice, every output group comes from at least one cart row, but keeping the guard makes the calculation safe and explicit.

Reusing the Existing Job Structure

The application structure is the same as Parts 8 and 9.

Initialize the SQL project:

oleander spark init part10-sql

Use:

• tutorial_part10_sql.py as the entrypoint file name
• tutorial as the custom library name

Then initialize the DataFrame project:

oleander spark init part10-dataframe

Use:

• tutorial_part10_dataframe.py as the entrypoint file name
• tutorial as the custom library name

Run uv sync in each project after initialization.

Both jobs accept the familiar date arguments:

• --data-interval-start
• --data-interval-end

They use UTC and a half-open interval, which means the start date is included and the end date is excluded.

We will focus on the new transformation in this part. The argparse and SparkSession setup remains the same as in the previous jobs.

The destination tables must already exist before the jobs run. The jobs transform and append data, but they do not create tables.

Building Cart Abandonment with Spark SQL

Here is the SQL transformation:

def build_daily_product_cart_abandonment(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    """
    Build daily product cart-abandonment rows for the given data interval.

    This function joins cart sessions with purchase sessions, summarizes the
    outcomes by event date and product, and inserts the result into
    oleander.tutorial.retail_daily_product_cart_abandonment.

    The output table is expected to already exist.
    """
    spark.sql(f"""
        INSERT INTO oleander.tutorial.retail_daily_product_cart_abandonment
        WITH filtered_session_products AS (
          SELECT *
          FROM oleander.tutorial.retail_session_products
          WHERE event_date >= CAST(CAST('{data_interval_start}' AS TIMESTAMP) AS DATE)
            AND event_date < CAST(CAST('{data_interval_end}' AS TIMESTAMP) AS DATE)
        ),
        cart_sessions AS (
          SELECT
            session_id,
            event_date,
            product_id,
            product_name,
            category
          FROM filtered_session_products
          WHERE added_to_cart
        ),
        purchase_sessions AS (
          SELECT
            session_id,
            product_id,
            TRUE AS purchased_cart
          FROM filtered_session_products
          WHERE purchased
        ),
        cart_outcomes AS (
          SELECT
            cart.event_date,
            cart.product_id,
            cart.product_name,
            cart.category,
            COALESCE(purchase.purchased_cart, FALSE) AS purchased_cart
          FROM cart_sessions AS cart
          LEFT JOIN purchase_sessions AS purchase
            ON cart.session_id = purchase.session_id
           AND cart.product_id = purchase.product_id
        ),
        cart_summary AS (
          SELECT
            event_date,
            product_id,
            product_name,
            category,
            COUNT(*) AS cart_session_count,
            SUM(CASE WHEN purchased_cart THEN 1 ELSE 0 END)
              AS purchased_cart_session_count,
            SUM(CASE WHEN NOT purchased_cart THEN 1 ELSE 0 END)
              AS abandoned_cart_session_count
          FROM cart_outcomes
          GROUP BY
            event_date,
            product_id,
            product_name,
            category
        )
        SELECT
          event_date,
          product_id,
          product_name,
          category,
          cart_session_count,
          purchased_cart_session_count,
          abandoned_cart_session_count,
          CAST(
            CASE
              WHEN cart_session_count > 0
                THEN abandoned_cart_session_count / cart_session_count
              ELSE 0
            END AS DECIMAL(9, 4)
          ) AS cart_abandonment_rate
        FROM cart_summary
    """)

This query uses several CTEs so that we can look at the transformation one step at a time.

Filtering the Source Rows

The filtered_session_products CTE limits the source table to the requested interval.

Both the cart and purchase datasets come from this filtered result. That keeps the job focused on one consistent batch of data.

Separating Cart and Purchase Sessions

The cart_sessions CTE keeps rows where added_to_cart is true. It also keeps the date and product details that we need in the final output.

The purchase_sessions CTE keeps rows where purchased is true. For these rows, we add a Boolean marker:

TRUE AS purchased_cart

This marker will tell us whether the join found a purchase.

Joining the Two Datasets

The cart_outcomes CTE places cart_sessions on the left and uses a left join:

LEFT JOIN purchase_sessions AS purchase
  ON cart.session_id = purchase.session_id
 AND cart.product_id = purchase.product_id

Every cart row remains in the result. If the same product was purchased in the same session, purchase.purchased_cart is true. If Spark cannot find a match, the value is null.

We convert that missing value to false with:

COALESCE(purchase.purchased_cart, FALSE)

COALESCE returns the first non-null value. So matched purchases stay true, while unmatched carts become false.

Notice that purchase-only rows do not appear in the result. The cart dataset is on the left, so the join begins with products that were actually added to a cart.

Aggregating the Outcomes

The cart_summary CTE groups the joined rows by date and product.

COUNT(*) gives us the total number of cart sessions. Conditional sums divide those rows into purchased and abandoned outcomes.

The outer SELECT then calculates the abandonment rate and places the columns in the same order as the destination table.

Building Cart Abandonment with the DataFrame API

Now let’s express the same transformation with the DataFrame API.

from pyspark.sql import functions as F


def build_daily_product_cart_abandonment(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    """
    Build daily product cart-abandonment rows for the given data interval.

    This function joins cart sessions with purchase sessions, summarizes the
    outcomes by event date and product, and inserts the result into
    oleander.tutorial.retail_daily_product_cart_abandonment_df.

    The output table is expected to already exist.
    """
    start_date = F.lit(data_interval_start).cast("timestamp").cast("date")
    end_date = F.lit(data_interval_end).cast("timestamp").cast("date")

    session_products = (
        spark.table("oleander.tutorial.retail_session_products_df")
        .filter(F.col("event_date") >= start_date)
        .filter(F.col("event_date") < end_date)
    )

    cart_sessions = (
        session_products
        .filter(F.col("added_to_cart"))
        .select(
            "session_id",
            "event_date",
            "product_id",
            "product_name",
            "category",
        )
    )

    purchase_sessions = (
        session_products
        .filter(F.col("purchased"))
        .select(
            "session_id",
            "product_id",
            F.lit(True).alias("purchased_cart"),
        )
    )

    cart_outcomes = (
        cart_sessions
        .join(
            purchase_sessions,
            ["session_id", "product_id"],
            "left",
        )
        .withColumn(
            "purchased_cart",
            F.coalesce(F.col("purchased_cart"), F.lit(False)),
        )
    )

    cart_summary = (
        cart_outcomes
        .groupBy(
            "event_date",
            "product_id",
            "product_name",
            "category",
        )
        .agg(
            F.count(F.lit(1)).alias("cart_session_count"),
            F.sum(
                F.when(F.col("purchased_cart"), F.lit(1))
                .otherwise(F.lit(0))
            ).alias("purchased_cart_session_count"),
            F.sum(
                F.when(~F.col("purchased_cart"), F.lit(1))
                .otherwise(F.lit(0))
            ).alias("abandoned_cart_session_count"),
        )
        .withColumn(
            "cart_abandonment_rate",
            F.when(
                F.col("cart_session_count") > F.lit(0),
                F.col("abandoned_cart_session_count")
                / F.col("cart_session_count"),
            )
            .otherwise(F.lit(0))
            .cast("decimal(9,4)"),
        )
        .select(
            "event_date",
            "product_id",
            "product_name",
            "category",
            "cart_session_count",
            "purchased_cart_session_count",
            "abandoned_cart_session_count",
            "cart_abandonment_rate",
        )
    )

    (
        cart_summary
        .writeTo("oleander.tutorial.retail_daily_product_cart_abandonment_df")
        .append()
    )

The DataFrame version follows the same stages as the SQL query.

We first create session_products for the requested interval. Then .filter() and .select() create the cart and purchase DataFrames.

The join is written like this:

.join(
    purchase_sessions,
    ["session_id", "product_id"],
    "left",
)

The list contains both join keys. The string "left" tells Spark to preserve every row from cart_sessions.

After the join, F.coalesce() changes a missing purchase marker from null to false.

Then .groupBy() and .agg() calculate the three session counts. The bitwise ~ operator negates the Boolean purchased_cart column, so it selects abandoned outcomes for the conditional sum.

Finally, .withColumn() calculates the rate, .select() aligns the output with the destination schema, and .writeTo().append() inserts the rows.

Reading the Results

After running either job, we can inspect the daily product metrics with:

spark.sql("""
    SELECT
      event_date,
      product_name,
      cart_session_count,
      purchased_cart_session_count,
      abandoned_cart_session_count,
      ROUND(cart_abandonment_rate * 100, 2) AS cart_abandonment_percent
    FROM oleander.tutorial.retail_daily_product_cart_abandonment
    ORDER BY
      event_date,
      cart_abandonment_rate DESC,
      cart_session_count DESC,
      product_name
""").show(truncate=False)

The counts give important context for the rate.

For example, a product with one cart session and one abandonment has a 100 percent abandonment rate, but that is only one session. A product with hundreds of cart sessions and a high abandonment rate may deserve more attention.

Comparing products can help us decide where to investigate, but the rate does not explain the cause by itself. Price, availability, shipping cost, checkout experience, comparison behavior, or tracking quality could all affect the result.

As always, a metric is evidence. It is not a tiny oracle living in the table.

Full Example Code

The full code for these jobs can be found here:

• SQL version
• DataFrame version

Deploying and Submitting the Jobs

The deployment process is the same as in the previous parts.

Build the artifacts from each project:

make

Upload the SQL job:

oleander spark jobs upload tutorial_part10_sql.py

Or upload the DataFrame job:

oleander spark jobs upload tutorial_part10_dataframe.py

Submit the SQL version for one day:

oleander spark jobs submit tutorial_part10_sql.py \
  --namespace tutorial \
  --name tutorial_part10_sql \
  --args=--data-interval-start=2026-01-01T00:00:00Z \
  --args=--data-interval-end=2026-01-02T00:00:00Z

Or submit the DataFrame version:

oleander spark jobs submit tutorial_part10_dataframe.py \
  --namespace tutorial \
  --name tutorial_part10_dataframe \
  --args=--data-interval-start=2026-01-01T00:00:00Z \
  --args=--data-interval-end=2026-01-02T00:00:00Z

These jobs still use append operations. If you run the same data interval again, they will append the same summary rows again and create duplicates.

So far, we have accepted that limitation and used fresh intervals. We will address it soon, but first we will pause and look more closely at the join concepts introduced in this part.

Summary

In this part, we measured product-level cart abandonment with joins.

We separated cart sessions from purchase sessions and joined them using both session_id and product_id. An inner join would have kept only completed carts, so we used a left join to preserve carts without a matching purchase.

We converted missing purchase matches into abandoned outcomes, then aggregated those outcomes into daily product counts and an abandonment rate.

We implemented the same logic in Spark SQL and the DataFrame API. Along the way, we introduced join keys, inner joins, left joins, nulls created by unmatched rows, and the importance of choosing the correct left-side dataset.

The jobs still have one operational weakness: running the same interval twice creates duplicate rows. We will address that in Part 12 with Iceberg’s MERGE INTO statement.

Before that, the next part will pause our practical pipeline work and build a clearer mental model of the different join types available in Spark.