# Measuring Cart Abandonment with Joins > Growing Tips (Apache Iceberg & Spark 101), part 10 of 16. > Some carts never ripen into purchases. Use joins to find what wilted on the vine and when. > Source: https://oleander.dev/grow/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: ```sql 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: | `session_id` | `product_id` | | ------------ | ------------ | | `session-1` | `product-a` | | `session-1` | `product-b` | | `session-2` | `product-a` | And our purchase dataset contains: | `session_id` | `product_id` | | ------------ | ------------ | | `session-1` | `product-a` | | `session-2` | `product-a` | 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: | `session_id` | `product_id` | | ------------ | ------------ | | `session-1` | `product-a` | | `session-2` | `product-a` | 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: | `session_id` | `product_id` | purchase matched | | ------------ | ------------ | ---------------- | | `session-1` | `product-a` | `true` | | `session-1` | `product-b` | `null` | | `session-2` | `product-a` | `true` | 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: ```text purchased cart sessions + abandoned cart sessions = total cart sessions ``` The abandonment rate is: ```text 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: ```python 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: ```python 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: ```bash 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: ```bash 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: ```python 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: ```sql 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: ```sql 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: ```sql 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. ```python 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: ```python .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: ```python 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](https://github.com/OleanderHQ/iceberg-spark-101-tutorials/blob/main/part10-sql/tutorial_part10_sql.py) - [DataFrame version](https://github.com/OleanderHQ/iceberg-spark-101-tutorials/blob/main/part10-dataframe/tutorial_part10_dataframe.py) ## Deploying and Submitting the Jobs The deployment process is the same as in the previous parts. Build the artifacts from each project: ```bash make ``` Upload the SQL job: ```bash oleander spark jobs upload tutorial_part10_sql.py ``` Or upload the DataFrame job: ```bash oleander spark jobs upload tutorial_part10_dataframe.py ``` Submit the SQL version for one day: ```bash 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: ```bash 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.