Ollie
In the previous part, we turned raw clickstream events into three kinds of summary tables. The most detailed one was retail_session_products, which gives us one row for each product that appeared in a user session.
That table is useful on its own, but it also gives us a foundation for answering more business-facing questions.
For example, how many sessions viewed a product? How many added it to the cart? How many reached checkout? And how many eventually purchased it?
In this part, we will use Spark aggregations to turn the session-product summary into a daily product funnel.
Just like in the previous part, we will write the job twice:
- once in Spark SQL
- once in the Spark DataFrame API
Both versions will produce the same result.
What Is a Product Funnel?
A product funnel is a way to summarize how users move through a sequence of actions related to a product.
For the fictional Oleander retail website, we will use these four stages:
- viewed the product
- added the product to the cart
- started checkout
- purchased the product
At a high level, we want to know how many sessions reached each stage.
If 100 sessions viewed a product, 30 added it to the cart, 15 started checkout, and 10 purchased it, the funnel gives us a compact view of where participation decreased.
It is important not to read too much into the shape of the funnel by itself. A smaller count tells us that fewer sessions reached a later stage, but it does not tell us why. The product may be too expensive. The checkout experience may be confusing. Users may simply be comparing products. The data shows us what happened, but understanding why usually takes more investigation.
It is also possible for events to be missing or to arrive in an unexpected order. A session might contain an add-to-cart event without a recorded product-view event. So the funnel is a useful behavioral summary, not proof that every user followed one perfect linear path through the website. Human beings are rarely that cooperative with our diagrams.
In this part, each output row will represent one product on one day.
Session Counts and Event Counts
Before creating the table, we need to distinguish two kinds of counts.
The first is a session count.
Suppose a user views the same product five times during one session. For the product-view stage of the funnel, that still represents one session that reached the stage.
So in this example:
- the product-view session count is
1 - the product-view event count is
5
For funnel conversion rates, we will use session counts. This prevents one user repeatedly performing the same action from making it look like more sessions entered the funnel.
But that does not mean event counts are useless.
Event counts tell us about repetition and intensity inside those sessions. Viewing the same product several times might suggest strong interest, comparison, or hesitation. Repeated add-to-cart or checkout events might be normal user behavior, a retry, or something worth checking in the event instrumentation.
A simple mental model is this:
- session counts tell us how many sessions reached a stage
- event counts tell us how many times the behavior happened
We will store both, but the product funnel itself will focus mainly on session counts.
Creating the Destination Tables
Let’s create the destination table for the SQL job from the interactive shell:
spark.sql(""" CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_daily_product_funnel ( event_date DATE,
product_id STRING, product_name STRING, category STRING,
product_view_session_count BIGINT, add_to_cart_session_count BIGINT, checkout_start_session_count BIGINT, purchase_session_count BIGINT,
product_view_event_count BIGINT, add_to_cart_event_count BIGINT, checkout_start_event_count BIGINT, purchase_event_count BIGINT,
revenue DECIMAL(18, 2),
view_to_cart_rate DECIMAL(9, 4), cart_to_checkout_rate DECIMAL(9, 4), checkout_to_purchase_rate DECIMAL(9, 4), view_to_purchase_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_funnel_df ( event_date DATE,
product_id STRING, product_name STRING, category STRING,
product_view_session_count BIGINT, add_to_cart_session_count BIGINT, checkout_start_session_count BIGINT, purchase_session_count BIGINT,
product_view_event_count BIGINT, add_to_cart_event_count BIGINT, checkout_start_event_count BIGINT, purchase_event_count BIGINT,
revenue DECIMAL(18, 2),
view_to_cart_rate DECIMAL(9, 4), cart_to_checkout_rate DECIMAL(9, 4), checkout_to_purchase_rate DECIMAL(9, 4), view_to_purchase_rate DECIMAL(9, 4) ) USING iceberg""")The first four columns identify the day and product. The next four count sessions at each funnel stage. After that, we store the corresponding raw event counts and total purchase revenue.
The final four columns are conversion rates:
view_to_cart_rateis add-to-cart sessions divided by product-view sessionscart_to_checkout_rateis checkout sessions divided by add-to-cart sessionscheckout_to_purchase_rateis purchase sessions divided by checkout sessionsview_to_purchase_rateis purchase sessions divided by product-view sessions
We store these rates as decimal values between 0 and 1 in the usual case. For example, 0.2500 means 25 percent.
If a denominator is zero, the job stores 0.0000. That avoids dividing by zero and gives the destination table a consistent numeric value.
Reusing the Part 8 Job Structure
The overall application structure is the same as in Part 8.
Initialize the SQL project:
oleander spark init part09-sqlUse:
tutorial_part09_sql.pyas the entrypoint file nametutorialas the custom library name
Then initialize the DataFrame project:
oleander spark init part09-dataframeUse:
tutorial_part09_dataframe.pyas the entrypoint file nametutorialas the custom library name
Run uv sync in each project after initialization.
Both jobs accept the same arguments we used in the previous part:
--data-interval-start--data-interval-end
They also use UTC and treat the interval as half-open: the start date is included, and the end date is excluded.
We will not repeat the complete argparse and SparkSession setup here. The structure is the same as Part 8. The important new work in this part is the product funnel transformation.
The destination tables must already exist before these jobs run. The jobs only transform and insert data; they do not create tables.
Building the Product Funnel with Spark SQL
Before looking at the complete transformation, we need to introduce one new SQL concept: a common table expression, usually shortened to CTE.
A CTE is a temporary named result that exists only while one SQL statement is running. It starts with WITH, followed by a name and the query that produces the result:
WITH some_name AS ( SELECT ...)SELECT *FROM some_nameYou can think of a CTE as giving an intermediate query result a useful name. It does not create a permanent table, and nothing remains after the statement finishes.
In our job, the CTE is named product_funnel. It performs the grouping and aggregation first. The outer SELECT can then use those aggregate column names to calculate the conversion rates. We could write the same logic as one more deeply nested query, but the CTE makes the two steps easier to see and discuss.
Here is the SQL transformation:
def build_daily_product_funnel( spark: SparkSession, data_interval_start: str, data_interval_end: str,) -> None: """ Build daily product funnel rows for the given data interval.
This function groups session-product summaries by event date and product, calculates session and event counts for each funnel stage, and inserts the result into oleander.tutorial.retail_daily_product_funnel.
The output table is expected to already exist. """ spark.sql(f""" INSERT INTO oleander.tutorial.retail_daily_product_funnel WITH product_funnel AS ( SELECT event_date, product_id, product_name, category,
SUM(CASE WHEN viewed_product THEN 1 ELSE 0 END) AS product_view_session_count, SUM(CASE WHEN added_to_cart THEN 1 ELSE 0 END) AS add_to_cart_session_count, SUM(CASE WHEN started_checkout THEN 1 ELSE 0 END) AS checkout_start_session_count, SUM(CASE WHEN purchased THEN 1 ELSE 0 END) AS purchase_session_count,
SUM(product_view_count) AS product_view_event_count, SUM(add_to_cart_count) AS add_to_cart_event_count, SUM(checkout_start_count) AS checkout_start_event_count, SUM(purchase_count) AS purchase_event_count,
CAST(SUM(revenue) AS DECIMAL(18, 2)) AS revenue 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) GROUP BY event_date, product_id, product_name, category ) SELECT event_date, product_id, product_name, category,
product_view_session_count, add_to_cart_session_count, checkout_start_session_count, purchase_session_count,
product_view_event_count, add_to_cart_event_count, checkout_start_event_count, purchase_event_count,
revenue,
CAST( CASE WHEN product_view_session_count > 0 THEN add_to_cart_session_count / product_view_session_count ELSE 0 END AS DECIMAL(9, 4) ) AS view_to_cart_rate, CAST( CASE WHEN add_to_cart_session_count > 0 THEN checkout_start_session_count / add_to_cart_session_count ELSE 0 END AS DECIMAL(9, 4) ) AS cart_to_checkout_rate, CAST( CASE WHEN checkout_start_session_count > 0 THEN purchase_session_count / checkout_start_session_count ELSE 0 END AS DECIMAL(9, 4) ) AS checkout_to_purchase_rate, CAST( CASE WHEN product_view_session_count > 0 THEN purchase_session_count / product_view_session_count ELSE 0 END AS DECIMAL(9, 4) ) AS view_to_purchase_rate FROM product_funnel """)There are two main parts to this query.
The first part is the product_funnel CTE. It filters the requested date interval and groups the source rows by:
event_dateproduct_idproduct_namecategory
Those grouping columns define the shape of the result. We get one row per day and product.
The session counts come from Boolean columns in retail_session_products.
For example:
SUM(CASE WHEN viewed_product THEN 1 ELSE 0 END)adds 1 for each session-product row where the product was viewed. Since Part 8 already reduced each session and product to one row, this gives us the number of sessions that viewed the product.
The event counts work differently:
SUM(product_view_count)Here, we preserve repeated behavior by adding the number of product-view events from every session-product row.
The second part of the query calculates conversion rates from the session counts created by the CTE.
For example:
add_to_cart_session_count / product_view_session_countcalculates the share of viewing sessions that also added the product to the cart.
Using an outer SELECT makes this easier to read because the conversion-rate expressions can refer to the aggregate column names from the CTE. Each expression also checks that its denominator is greater than zero before dividing.
Building the Product Funnel with the DataFrame API
Now let’s build the same product funnel with the DataFrame API.
from pyspark.sql import functions as F
def build_daily_product_funnel( spark: SparkSession, data_interval_start: str, data_interval_end: str,) -> None: """ Build daily product funnel rows for the given data interval.
This function groups session-product summaries by event date and product, calculates session and event counts for each funnel stage, and inserts the result into oleander.tutorial.retail_daily_product_funnel_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")
product_funnel = ( spark.table("oleander.tutorial.retail_session_products_df") .filter(F.col("event_date") >= start_date) .filter(F.col("event_date") < end_date) .groupBy( "event_date", "product_id", "product_name", "category", ) .agg( F.sum( F.when(F.col("viewed_product"), F.lit(1)) .otherwise(F.lit(0)) ).alias("product_view_session_count"), F.sum( F.when(F.col("added_to_cart"), F.lit(1)) .otherwise(F.lit(0)) ).alias("add_to_cart_session_count"), F.sum( F.when(F.col("started_checkout"), F.lit(1)) .otherwise(F.lit(0)) ).alias("checkout_start_session_count"), F.sum( F.when(F.col("purchased"), F.lit(1)) .otherwise(F.lit(0)) ).alias("purchase_session_count"), F.sum("product_view_count").alias("product_view_event_count"), F.sum("add_to_cart_count").alias("add_to_cart_event_count"), F.sum("checkout_start_count").alias("checkout_start_event_count"), F.sum("purchase_count").alias("purchase_event_count"), F.sum("revenue").cast("decimal(18,2)").alias("revenue"), ) .withColumn( "view_to_cart_rate", F.when( F.col("product_view_session_count") > F.lit(0), F.col("add_to_cart_session_count") / F.col("product_view_session_count"), ) .otherwise(F.lit(0)) .cast("decimal(9,4)"), ) .withColumn( "cart_to_checkout_rate", F.when( F.col("add_to_cart_session_count") > F.lit(0), F.col("checkout_start_session_count") / F.col("add_to_cart_session_count"), ) .otherwise(F.lit(0)) .cast("decimal(9,4)"), ) .withColumn( "checkout_to_purchase_rate", F.when( F.col("checkout_start_session_count") > F.lit(0), F.col("purchase_session_count") / F.col("checkout_start_session_count"), ) .otherwise(F.lit(0)) .cast("decimal(9,4)"), ) .withColumn( "view_to_purchase_rate", F.when( F.col("product_view_session_count") > F.lit(0), F.col("purchase_session_count") / F.col("product_view_session_count"), ) .otherwise(F.lit(0)) .cast("decimal(9,4)"), ) .select( "event_date", "product_id", "product_name", "category", "product_view_session_count", "add_to_cart_session_count", "checkout_start_session_count", "purchase_session_count", "product_view_event_count", "add_to_cart_event_count", "checkout_start_event_count", "purchase_event_count", "revenue", "view_to_cart_rate", "cart_to_checkout_rate", "checkout_to_purchase_rate", "view_to_purchase_rate", ) )
( product_funnel .writeTo("oleander.tutorial.retail_daily_product_funnel_df") .append() )The structure is the same as the SQL version.
We start with retail_session_products_df, filter the date interval, and call .groupBy() with the columns that define one output row.
Inside .agg(), F.when() plays the same role as CASE WHEN in SQL. We use it to turn each Boolean behavior flag into 1 or 0, then sum those values to get session counts.
The existing event-count columns and revenue can be summed directly.
After the aggregation, each .withColumn() calculates one conversion rate. Just like the SQL version, the expression checks the denominator before dividing and casts the result to decimal(9,4).
Finally, .select() places the columns in the same order as the destination table, and .writeTo().append() inserts the result.
Reading the Product Funnel
After running either job, we can inspect the session funnel with a query like this:
spark.sql(""" SELECT event_date, product_name, product_view_session_count, add_to_cart_session_count, checkout_start_session_count, purchase_session_count, revenue, ROUND(view_to_cart_rate * 100, 2) AS view_to_cart_percent, ROUND(cart_to_checkout_rate * 100, 2) AS cart_to_checkout_percent, ROUND(checkout_to_purchase_rate * 100, 2) AS checkout_to_purchase_percent, ROUND(view_to_purchase_rate * 100, 2) AS view_to_purchase_percent FROM oleander.tutorial.retail_daily_product_funnel ORDER BY event_date, product_view_session_count DESC, product_name""").show(truncate=False)The session counts show how many sessions reached each stage. The conversion rates make it easier to compare products even when their traffic volumes are different.
For example, one product may have many viewing sessions but a relatively low view-to-cart rate. Another may have fewer viewing sessions but a much larger share of those sessions eventually purchase.
Those patterns tell us where to investigate. They do not explain the cause by themselves. A funnel is a starting point for asking better questions, not a machine that automatically produces business wisdom.
Also remember that the counts may not always decrease perfectly at each stage. Events can be missing, users can enter a flow in unexpected ways, and real tracking data is not always as tidy as a tutorial diagram.
Using Event Counts to Understand Repetition
Now let’s look at what event counts can tell us.
One useful measure is the average number of product-view events per viewing session:
spark.sql(""" SELECT event_date, product_name, product_view_session_count, product_view_event_count, CASE WHEN product_view_session_count > 0 THEN ROUND( product_view_event_count / product_view_session_count, 2 ) ELSE 0 END AS views_per_viewing_session FROM oleander.tutorial.retail_daily_product_funnel ORDER BY event_date, views_per_viewing_session DESC, product_name""").show(truncate=False)If product_view_session_count is 10 and product_view_event_count is 30, then the product had an average of three view events per viewing session.
That does not mean 30 separate sessions were interested in the product. It means the 10 sessions generated 30 view events in total.
A larger value may indicate that users returned to the product several times. That could mean strong interest. It could mean uncertainty. It could also mean that the website or event tracking produced repeated events. The number gives us a clue, but its meaning still depends on the product, the website experience, and how events are generated.
The same idea applies to the other event counts. Repeated add-to-cart, checkout, or purchase events may represent legitimate repeated actions, retries, or a data-quality issue worth investigating.
This is why it is useful to keep session counts and event counts separate. They describe related behavior, but they do not answer the same question.
Full Example Code
The full code for these jobs can be found here:
Deploying and Submitting the Jobs
The deployment process is the same as in Part 8.
Build the deployable artifacts from each project:
makeUpload the SQL job:
oleander spark jobs upload tutorial_part09_sql.pyOr upload the DataFrame job:
oleander spark jobs upload tutorial_part09_dataframe.pyTo submit the SQL version for one day:
oleander spark jobs submit tutorial_part09_sql.py \ --namespace tutorial \ --name tutorial_part09_sql \ --args=--data-interval-start=2026-01-01T00:00:00Z \ --args=--data-interval-end=2026-01-02T00:00:00ZOr submit the DataFrame version:
oleander spark jobs submit tutorial_part09_dataframe.py \ --namespace tutorial \ --name tutorial_part09_dataframe \ --args=--data-interval-start=2026-01-01T00:00:00Z \ --args=--data-interval-end=2026-01-02T00:00:00ZFor now, use a date interval that has not already been processed. These jobs append rows to the destination tables, so running the same interval again will create duplicates.
Later in this tutorial series, we will look at how to make jobs safer to retry. For the moment, it is enough to be aware of the limitation.
Summary
In this part, we used Spark aggregations to turn session-product summaries into daily product funnels.
We counted how many sessions viewed each product, added it to the cart, started checkout, and purchased it. We then used those session counts to calculate conversion rates between funnel stages.
We also kept the raw event counts. Session counts tell us how many sessions reached a stage, while event counts help us understand how often users repeated the behavior inside those sessions.
We implemented the same transformation in Spark SQL and the DataFrame API, and both versions wrote the result to Iceberg tables.
The funnel gives us a broad view of where participation decreases. In the next part, we will ask a more specific question: which products were added to a cart but never purchased? To answer that, we will introduce joins.