Summarizing Clickstream Events with Spark

[8]of 16

Ollie

head gardener

In the previous part, we created the retail_clickstream_events table, which contains user clickstream events from the fictional Oleander retail website.

From this part, we will move from raw event data to summary tables.

This is an important step in practical analytics work. Raw events are useful, but they are often too detailed to understand directly. To answer real business questions, we usually build more structured summaries on top of them.

In this part, we will write two equivalent versions of the same jobs:

one in Spark SQL
one in the Spark DataFrame API

Both approaches work, and both produce the same results. So why learn both?

If you already have experience with databases, SQL may feel more natural. The DataFrame API is more programmatic and is often easier to turn into reusable application code. In practice, it is worth being comfortable with both.

What We Are Going to Build

We will build three kinds of summary tables:

a daily traffic summary
a session summary
a session-product summary

We will also create two versions of each destination table:

one for the jobs written in Spark SQL
one for the jobs written in the DataFrame API

That way, we can compare the two styles directly without mixing their outputs.

Creating the Destination Tables

Let’s start by creating the destination tables from the interactive shell.

Note Feel free to skim the DDL below and focus mainly on the shape of the destination tables. The important thing for now is to understand what kinds of summary tables we are building, not to memorize every column definition.

Daily traffic summary table

python

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

      total_events BIGINT,
      unique_users BIGINT,
      unique_sessions BIGINT,

      page_view_count BIGINT,
      search_count BIGINT,
      product_view_count BIGINT,
      add_to_cart_count BIGINT,
      remove_from_cart_count BIGINT,
      checkout_start_count BIGINT,
      purchase_count BIGINT
    )
    USING iceberg
""")

Session summary table

python

spark.sql("""
    CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_sessions (
      session_id STRING,
      user_id STRING,

      session_date DATE,
      session_start_time TIMESTAMP,
      session_end_time TIMESTAMP,
      session_duration_seconds BIGINT,

      event_count BIGINT,

      page_view_count BIGINT,
      search_count BIGINT,
      product_view_count BIGINT,
      add_to_cart_count BIGINT,
      remove_from_cart_count BIGINT,
      checkout_start_count BIGINT,
      purchase_count BIGINT,

      searched BOOLEAN,
      viewed_product BOOLEAN,
      added_to_cart BOOLEAN,
      started_checkout BOOLEAN,
      purchased BOOLEAN
    )
    USING iceberg
""")

Session-product summary table

python

spark.sql("""
    CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_session_products (
      session_id STRING,
      user_id STRING,

      event_date DATE,

      product_id STRING,
      product_name STRING,
      category STRING,

      first_product_view_time TIMESTAMP,
      first_add_to_cart_time TIMESTAMP,
      first_remove_from_cart_time TIMESTAMP,
      first_checkout_start_time TIMESTAMP,
      first_purchase_time TIMESTAMP,

      product_view_count BIGINT,
      add_to_cart_count BIGINT,
      remove_from_cart_count BIGINT,
      checkout_start_count BIGINT,
      purchase_count BIGINT,

      viewed_product BOOLEAN,
      added_to_cart BOOLEAN,
      removed_from_cart BOOLEAN,
      started_checkout BOOLEAN,
      purchased BOOLEAN,

      quantity_added BIGINT,
      quantity_purchased BIGINT,

      revenue DECIMAL(12, 2)
    )
    USING iceberg
""")

Now let’s create the same three destination tables for the DataFrame version.

Daily traffic summary table for DataFrame API

python

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

      total_events BIGINT,
      unique_users BIGINT,
      unique_sessions BIGINT,

      page_view_count BIGINT,
      search_count BIGINT,
      product_view_count BIGINT,
      add_to_cart_count BIGINT,
      remove_from_cart_count BIGINT,
      checkout_start_count BIGINT,
      purchase_count BIGINT
    )
    USING iceberg
""")

Session summary table for DataFrame API

python

spark.sql("""
    CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_sessions_df (
      session_id STRING,
      user_id STRING,

      session_date DATE,
      session_start_time TIMESTAMP,
      session_end_time TIMESTAMP,
      session_duration_seconds BIGINT,

      event_count BIGINT,

      page_view_count BIGINT,
      search_count BIGINT,
      product_view_count BIGINT,
      add_to_cart_count BIGINT,
      remove_from_cart_count BIGINT,
      checkout_start_count BIGINT,
      purchase_count BIGINT,

      searched BOOLEAN,
      viewed_product BOOLEAN,
      added_to_cart BOOLEAN,
      started_checkout BOOLEAN,
      purchased BOOLEAN
    )
    USING iceberg
""")

Session-product summary table for DataFrame API

python

spark.sql("""
    CREATE TABLE IF NOT EXISTS oleander.tutorial.retail_session_products_df (
      session_id STRING,
      user_id STRING,

      event_date DATE,

      product_id STRING,
      product_name STRING,
      category STRING,

      first_product_view_time TIMESTAMP,
      first_add_to_cart_time TIMESTAMP,
      first_remove_from_cart_time TIMESTAMP,
      first_checkout_start_time TIMESTAMP,
      first_purchase_time TIMESTAMP,

      product_view_count BIGINT,
      add_to_cart_count BIGINT,
      remove_from_cart_count BIGINT,
      checkout_start_count BIGINT,
      purchase_count BIGINT,

      viewed_product BOOLEAN,
      added_to_cart BOOLEAN,
      removed_from_cart BOOLEAN,
      started_checkout BOOLEAN,
      purchased BOOLEAN,

      quantity_added BIGINT,
      quantity_purchased BIGINT,

      revenue DECIMAL(12, 2)
    )
    USING iceberg
""")

As you can see, we created two sets of three tables. One set is for jobs written in Spark SQL, and the other is for jobs written in the Spark DataFrame API.

Understanding the Summary Tables

Before we write the jobs, it helps to be clear about what each table is supposed to represent.

`retail_daily_traffic_summary`

This table summarizes overall traffic at the day level.

It tells us:

how many events happened on each day
how many unique users visited
how many unique sessions there were
how many times each event type occurred

This is the broadest summary in this part.

`retail_sessions`

This table summarizes behavior at the session level.

When a user visits the website, that visit belongs to a session. This table describes what happened during each session: when it started, when it ended, how many events occurred, and whether certain actions happened during that session.

This gives us a more detailed view than the daily summary.

`retail_session_products`

This table goes one step deeper and summarizes activity at the session-and-product level.

A user may visit the website just to browse, or they may be interested in a specific product. We can never know their exact intention, but we can still summarize their behavior for each product within each session.

That makes this table useful for understanding product-level engagement inside user sessions.

Initializing the Projects

As mentioned earlier, we will write two equivalent jobs: one in SQL and one in the DataFrame API.

Let’s initialize a Spark project for each one.

For the SQL version:

bash

oleander spark init part08-sql

Use:

tutorial_part08_sql.py as the entrypoint file name
tutorial as the custom library name

We are not going to use the custom library much here, so the exact package name does not matter very much.

Now initialize the DataFrame version:

bash

oleander spark init part08-dataframe

Use:

tutorial_part08_dataframe.py as the entrypoint file name
tutorial as the custom library name

As explained in the previous parts, these are uv-managed Python projects. After initialization, run:

bash

uv sync

and activate the virtual environment as needed.

Accepting a Date Range as Arguments

In real analytics work, we usually do not run large summary jobs over an entire table every time, especially when the source data is time-series data.

Our source table contains retail clickstream events, which naturally grow over time. Instead of summarizing the whole table every day, we usually build summaries incrementally.

For example, suppose today is 2026-01-06, and we already built summaries for 2026-01-01 through 2026-01-04. Then we would usually want to summarize only 2026-01-05, because the earlier summaries should not change.

We do not want to rewrite the job every day just to change the date. Instead, we pass the data interval as command-line arguments.

That part is not really specific to Spark. It is just normal Python program structure. This is worth remembering: your Spark application is still just a Python program until it starts interacting with Spark through SparkSession.

Here is the basic setup:

python

import argparse
from pyspark.sql import SparkSession


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-interval-start", required=True)
    parser.add_argument("--data-interval-end", required=True)
    return parser.parse_args()


def main():
    args = parse_args()

    spark = (
        SparkSession.builder
        .appName("tutorial-part08-sql")
        .getOrCreate()
    )

    spark.conf.set("spark.sql.session.timeZone", "UTC")


if __name__ == "__main__":
    main()

This uses Python’s standard argparse library and nothing Spark-specific yet beyond creating the SparkSession.

One line is especially worth calling out:

python

spark.conf.set("spark.sql.session.timeZone", "UTC")

This tells Spark to use UTC when interpreting and working with timestamps.

That matters because time zones can easily make analytics results confusing. Fortunately, or unfortunately, we live on a spherical Earth orbiting around the Sun, which means different places experience different local times. For convenience, people defined time zones. That is helpful for daily life, because it tells you what time it is where you are. But in data analysis, time zones often create confusion.

Some users might visit the Oleander retail website from Seattle while others visit from South Korea. Even if they visit at exactly the same moment, their local clock times are different. We do not want that difference to make the data harder to reason about. That is why we standardize on UTC.

UTC stands for Coordinated Universal Time. I still do not know why it is not called CUT, but either way, it is the main time standard that other time zones are defined relative to.

Building a Daily Traffic Summary with SQL

Let’s start with the daily traffic summary.

We want to know:

how many total events happened each day
how many unique users visited
how many unique sessions there were
how many events of each type occurred

Here is one way to do that in Spark SQL:

python

def build_daily_traffic_summary(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    spark.sql(f"""
        INSERT INTO oleander.tutorial.retail_daily_traffic_summary
        SELECT
          CAST(event_time AS DATE) AS event_date,
          COUNT(*) AS total_events,
          COUNT(DISTINCT user_id) AS unique_users,
          COUNT(DISTINCT session_id) AS unique_sessions,
          SUM(CASE WHEN event_type = 'page_view' THEN 1 ELSE 0 END) AS page_view_count,
          SUM(CASE WHEN event_type = 'search' THEN 1 ELSE 0 END) AS search_count,
          SUM(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) AS product_view_count,
          SUM(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS add_to_cart_count,
          SUM(CASE WHEN event_type = 'remove_from_cart' THEN 1 ELSE 0 END) AS remove_from_cart_count,
          SUM(CASE WHEN event_type = 'checkout_start' THEN 1 ELSE 0 END) AS checkout_start_count,
          SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS purchase_count
        FROM oleander.tutorial.retail_clickstream_events
        WHERE event_time >= TIMESTAMP '{data_interval_start}'
          AND event_time < TIMESTAMP '{data_interval_end}'
        GROUP BY CAST(event_time AS DATE)
    """)

This query introduces some very common SQL patterns:

WHERE filters the rows before aggregation
GROUP BY defines the granularity of the result
COUNT and SUM are aggregate functions
CASE WHEN ... THEN ... ELSE ... END lets us compute conditional counts

For example, this expression:

sql

SUM(CASE WHEN event_type = 'page_view' THEN 1 ELSE 0 END)

counts only the rows where event_type is page_view.

That pattern shows up all the time in analytics queries.

If you want to explore more built-in SQL functions, Spark’s SQL function reference is here: https://spark.apache.org/docs/3.5.6/api/sql/index.html

Building a Daily Traffic Summary with the DataFrame API

SQL is great, but it is also worth learning the DataFrame API. Here is the DataFrame version of the same daily traffic summary job:

python

from pyspark.sql import functions as F


def build_daily_traffic_summary(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    (
        spark.table("oleander.tutorial.retail_clickstream_events")
        .filter(F.col("event_time") >= F.lit(data_interval_start).cast("timestamp"))
        .filter(F.col("event_time") < F.lit(data_interval_end).cast("timestamp"))
        .groupBy(F.col("event_time").cast("date").alias("event_date"))
        .agg(
            F.count(F.lit(1)).alias("total_events"),
            F.countDistinct("user_id").alias("unique_users"),
            F.countDistinct("session_id").alias("unique_sessions"),
            F.sum(
                F.when(F.col("event_type") == "page_view", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("page_view_count"),
            F.sum(
                F.when(F.col("event_type") == "search", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("search_count"),
            F.sum(
                F.when(F.col("event_type") == "product_view", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("product_view_count"),
            F.sum(
                F.when(F.col("event_type") == "add_to_cart", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("add_to_cart_count"),
            F.sum(
                F.when(F.col("event_type") == "remove_from_cart", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("remove_from_cart_count"),
            F.sum(
                F.when(F.col("event_type") == "checkout_start", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("checkout_start_count"),
            F.sum(
                F.when(F.col("event_type") == "purchase", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("purchase_count"),
        )
        .select(
            "event_date",
            "total_events",
            "unique_users",
            "unique_sessions",
            "page_view_count",
            "search_count",
            "product_view_count",
            "add_to_cart_count",
            "remove_from_cart_count",
            "checkout_start_count",
            "purchase_count",
        )
        .writeTo("oleander.tutorial.retail_daily_traffic_summary_df")
        .append()
    )

Conceptually, the DataFrame version applies a chain of operations to a DataFrame:

start with the source table
filter rows
group them
aggregate them
select the output columns
write the result

One important Spark concept shows up clearly here: lazy execution.

Most DataFrame operations such as .filter(), .groupBy(), .agg(), and .select() do not immediately process data. Instead, they build an internal plan. The actual work starts only when Spark reaches an action.

In this example, the action is:

python

.writeTo("oleander.tutorial.retail_daily_traffic_summary_df").append()

That is when Spark actually executes the plan.

As a beginner, the most important thing to remember is this:

transformations describe what should happen
actions actually trigger execution

This lazy model gives Spark room to optimize the overall plan before it starts running it.

Building a Session Summary with SQL

Now let’s build a session summary.

The daily summary groups by date. This time, we group by session_id and user_id so that we get one row per user session.

python

def build_sessions(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    spark.sql(f"""
        INSERT INTO oleander.tutorial.retail_sessions
        SELECT
          session_id,
          user_id,
          CAST(MIN(event_time) AS DATE) AS session_date,
          MIN(event_time) AS session_start_time,
          MAX(event_time) AS session_end_time,
          unix_timestamp(MAX(event_time)) - unix_timestamp(MIN(event_time))
            AS session_duration_seconds,
          COUNT(*) AS event_count,
          SUM(CASE WHEN event_type = 'page_view' THEN 1 ELSE 0 END) AS page_view_count,
          SUM(CASE WHEN event_type = 'search' THEN 1 ELSE 0 END) AS search_count,
          SUM(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) AS product_view_count,
          SUM(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS add_to_cart_count,
          SUM(CASE WHEN event_type = 'remove_from_cart' THEN 1 ELSE 0 END) AS remove_from_cart_count,
          SUM(CASE WHEN event_type = 'checkout_start' THEN 1 ELSE 0 END) AS checkout_start_count,
          SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS purchase_count,
          SUM(CASE WHEN event_type = 'search' THEN 1 ELSE 0 END) > 0 AS searched,
          SUM(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) > 0 AS viewed_product,
          SUM(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) > 0 AS added_to_cart,
          SUM(CASE WHEN event_type = 'checkout_start' THEN 1 ELSE 0 END) > 0 AS started_checkout,
          SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) > 0 AS purchased
        FROM oleander.tutorial.retail_clickstream_events
        WHERE event_time >= TIMESTAMP '{data_interval_start}'
          AND event_time < TIMESTAMP '{data_interval_end}'
        GROUP BY
          session_id,
          user_id
    """)

Syntactically, this is not very different from the daily summary query. The main change is the grouping key and the set of output columns.

This is also a good place to clarify something about spark.sql().

In the DataFrame API, the separation between transformation and action is explicit. In SQL, it is a little more implicit.

For example:

python

df = spark.sql("SELECT * FROM oleander.tutorial.retail_daily_traffic_summary")

This returns a DataFrame, but it does not immediately read all the data. Spark builds a plan first. When you later call an action such as .show() or .collect(), the query actually executes.

But some SQL statements do execute immediately. For example, INSERT INTO changes data, so Spark performs that operation right away.

A simple mental model is:

pure query statements such as SELECT are lazy
statements with side effects such as INSERT INTO execute immediately

That is the same underlying idea you already saw in the DataFrame API, just expressed through SQL.

Building a Session Summary with the DataFrame API

Now let’s write the same session summary using the DataFrame API:

python

def build_sessions(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    (
        spark.table("oleander.tutorial.retail_clickstream_events")
        .filter(F.col("event_time") >= F.lit(data_interval_start).cast("timestamp"))
        .filter(F.col("event_time") < F.lit(data_interval_end).cast("timestamp"))
        .groupBy("session_id", "user_id")
        .agg(
            F.min("event_time").cast("date").alias("session_date"),
            F.min("event_time").alias("session_start_time"),
            F.max("event_time").alias("session_end_time"),
            (
                F.unix_timestamp(F.max("event_time"))
                - F.unix_timestamp(F.min("event_time"))
            ).alias("session_duration_seconds"),
            F.count(F.lit(1)).alias("event_count"),
            F.sum(
                F.when(F.col("event_type") == "page_view", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("page_view_count"),
            F.sum(
                F.when(F.col("event_type") == "search", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("search_count"),
            F.sum(
                F.when(F.col("event_type") == "product_view", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("product_view_count"),
            F.sum(
                F.when(F.col("event_type") == "add_to_cart", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("add_to_cart_count"),
            F.sum(
                F.when(F.col("event_type") == "remove_from_cart", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("remove_from_cart_count"),
            F.sum(
                F.when(F.col("event_type") == "checkout_start", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("checkout_start_count"),
            F.sum(
                F.when(F.col("event_type") == "purchase", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("purchase_count"),
            (
                F.sum(
                    F.when(F.col("event_type") == "search", F.lit(1))
                    .otherwise(F.lit(0))
                )
                > F.lit(0)
            ).alias("searched"),
            (
                F.sum(
                    F.when(F.col("event_type") == "product_view", F.lit(1))
                    .otherwise(F.lit(0))
                )
                > F.lit(0)
            ).alias("viewed_product"),
            (
                F.sum(
                    F.when(F.col("event_type") == "add_to_cart", F.lit(1))
                    .otherwise(F.lit(0))
                )
                > F.lit(0)
            ).alias("added_to_cart"),
            (
                F.sum(
                    F.when(F.col("event_type") == "checkout_start", F.lit(1))
                    .otherwise(F.lit(0))
                )
                > F.lit(0)
            ).alias("started_checkout"),
            (
                F.sum(
                    F.when(F.col("event_type") == "purchase", F.lit(1))
                    .otherwise(F.lit(0))
                )
                > F.lit(0)
            ).alias("purchased"),
        )
        .select(
            "session_id",
            "user_id",
            "session_date",
            "session_start_time",
            "session_end_time",
            "session_duration_seconds",
            "event_count",
            "page_view_count",
            "search_count",
            "product_view_count",
            "add_to_cart_count",
            "remove_from_cart_count",
            "checkout_start_count",
            "purchase_count",
            "searched",
            "viewed_product",
            "added_to_cart",
            "started_checkout",
            "purchased",
        )
        .writeTo("oleander.tutorial.retail_sessions_df")
        .append()
    )

This is the same job, just expressed in a different style.

Building a Session-Product Summary with SQL

The next summary goes even more granular. We now want one row per session, user, and product.

python

def build_session_products(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    spark.sql(f"""
        INSERT INTO oleander.tutorial.retail_session_products
        SELECT
          session_id,
          user_id,

          CAST(MIN(event_time) AS DATE) AS event_date,

          product_id,
          product_name,
          category,

          MIN(CASE WHEN event_type = 'product_view' THEN event_time END) AS first_product_view_time,
          MIN(CASE WHEN event_type = 'add_to_cart' THEN event_time END) AS first_add_to_cart_time,
          MIN(CASE WHEN event_type = 'remove_from_cart' THEN event_time END) AS first_remove_from_cart_time,
          MIN(CASE WHEN event_type = 'checkout_start' THEN event_time END) AS first_checkout_start_time,
          MIN(CASE WHEN event_type = 'purchase' THEN event_time END) AS first_purchase_time,

          SUM(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) AS product_view_count,
          SUM(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS add_to_cart_count,
          SUM(CASE WHEN event_type = 'remove_from_cart' THEN 1 ELSE 0 END) AS remove_from_cart_count,
          SUM(CASE WHEN event_type = 'checkout_start' THEN 1 ELSE 0 END) AS checkout_start_count,
          SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS purchase_count,

          SUM(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) > 0 AS viewed_product,
          SUM(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) > 0 AS added_to_cart,
          SUM(CASE WHEN event_type = 'remove_from_cart' THEN 1 ELSE 0 END) > 0 AS removed_from_cart,
          SUM(CASE WHEN event_type = 'checkout_start' THEN 1 ELSE 0 END) > 0 AS started_checkout,
          SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) > 0 AS purchased,

          SUM(
            CASE
              WHEN event_type = 'add_to_cart' THEN COALESCE(quantity, 0)
              ELSE 0
            END
          ) AS quantity_added,

          SUM(
            CASE
              WHEN event_type = 'purchase' THEN COALESCE(quantity, 0)
              ELSE 0
            END
          ) AS quantity_purchased,

          CAST(
            SUM(
              CASE
                WHEN event_type = 'purchase'
                  THEN COALESCE(quantity, 0) * COALESCE(price, CAST(0.00 AS DECIMAL(10, 2)))
                ELSE CAST(0.00 AS DECIMAL(10, 2))
              END
            ) AS DECIMAL(12, 2)
          ) AS revenue

        FROM oleander.tutorial.retail_clickstream_events
        WHERE event_time >= TIMESTAMP '{data_interval_start}'
          AND event_time < TIMESTAMP '{data_interval_end}'
          AND product_id IS NOT NULL
        GROUP BY
          session_id,
          user_id,
          product_id,
          product_name,
          category
    """)

This query adds a few useful analytics ideas:

first event time per event type
product-level behavior flags
quantity totals
revenue totals

That is already enough to support more interesting analysis in the next parts.

Building a Session-Product Summary with the DataFrame API

And here is the DataFrame version:

python

def build_session_products(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    (
        spark.table("oleander.tutorial.retail_clickstream_events")
        .filter(F.col("event_time") >= F.lit(data_interval_start).cast("timestamp"))
        .filter(F.col("event_time") < F.lit(data_interval_end).cast("timestamp"))
        .filter(F.col("product_id").isNotNull())
        .groupBy(
            "session_id",
            "user_id",
            "product_id",
            "product_name",
            "category",
        )
        .agg(
            F.min("event_time").cast("date").alias("event_date"),
            F.min(
                F.when(F.col("event_type") == "product_view", F.col("event_time"))
            ).alias("first_product_view_time"),
            F.min(
                F.when(F.col("event_type") == "add_to_cart", F.col("event_time"))
            ).alias("first_add_to_cart_time"),
            F.min(
                F.when(F.col("event_type") == "remove_from_cart", F.col("event_time"))
            ).alias("first_remove_from_cart_time"),
            F.min(
                F.when(F.col("event_type") == "checkout_start", F.col("event_time"))
            ).alias("first_checkout_start_time"),
            F.min(
                F.when(F.col("event_type") == "purchase", F.col("event_time"))
            ).alias("first_purchase_time"),
            F.sum(
                F.when(F.col("event_type") == "product_view", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("product_view_count"),
            F.sum(
                F.when(F.col("event_type") == "add_to_cart", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("add_to_cart_count"),
            F.sum(
                F.when(F.col("event_type") == "remove_from_cart", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("remove_from_cart_count"),
            F.sum(
                F.when(F.col("event_type") == "checkout_start", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("checkout_start_count"),
            F.sum(
                F.when(F.col("event_type") == "purchase", F.lit(1))
                .otherwise(F.lit(0))
            ).alias("purchase_count"),
            (
                F.sum(
                    F.when(F.col("event_type") == "product_view", F.lit(1))
                    .otherwise(F.lit(0))
                ) > F.lit(0)
            ).alias("viewed_product"),
            (
                F.sum(
                    F.when(F.col("event_type") == "add_to_cart", F.lit(1))
                    .otherwise(F.lit(0))
                ) > F.lit(0)
            ).alias("added_to_cart"),
            (
                F.sum(
                    F.when(F.col("event_type") == "remove_from_cart", F.lit(1))
                    .otherwise(F.lit(0))
                ) > F.lit(0)
            ).alias("removed_from_cart"),
            (
                F.sum(
                    F.when(F.col("event_type") == "checkout_start", F.lit(1))
                    .otherwise(F.lit(0))
                ) > F.lit(0)
            ).alias("started_checkout"),
            (
                F.sum(
                    F.when(F.col("event_type") == "purchase", F.lit(1))
                    .otherwise(F.lit(0))
                ) > F.lit(0)
            ).alias("purchased"),
            F.sum(
                F.when(
                    F.col("event_type") == "add_to_cart",
                    F.coalesce(F.col("quantity"), F.lit(0)),
                ).otherwise(F.lit(0))
            ).alias("quantity_added"),
            F.sum(
                F.when(
                    F.col("event_type") == "purchase",
                    F.coalesce(F.col("quantity"), F.lit(0)),
                ).otherwise(F.lit(0))
            ).alias("quantity_purchased"),
            F.sum(
                F.when(
                    F.col("event_type") == "purchase",
                    F.coalesce(F.col("quantity"), F.lit(0))
                    * F.coalesce(F.col("price"), F.lit("0.00").cast("decimal(10,2)")),
                ).otherwise(F.lit("0.00").cast("decimal(10,2)"))
            ).cast("decimal(12,2)").alias("revenue"),
        )
        .select(
            "session_id",
            "user_id",
            "event_date",
            "product_id",
            "product_name",
            "category",
            "first_product_view_time",
            "first_add_to_cart_time",
            "first_remove_from_cart_time",
            "first_checkout_start_time",
            "first_purchase_time",
            "product_view_count",
            "add_to_cart_count",
            "remove_from_cart_count",
            "checkout_start_count",
            "purchase_count",
            "viewed_product",
            "added_to_cart",
            "removed_from_cart",
            "started_checkout",
            "purchased",
            "quantity_added",
            "quantity_purchased",
            "revenue",
        )
        .writeTo("oleander.tutorial.retail_session_products_df")
        .append()
    )

Full Example Code

The full code for these jobs can be found here:

Deploying and Submitting the Jobs

At this point, you have written your first real Spark jobs. That is a good milestone.

With Oleander, deploying and submitting a job is straightforward.

First, build the deployable artifacts:

bash

make

Then upload the SQL version:

bash

oleander spark jobs upload tutorial_part08_sql.py

or upload the DataFrame version:

bash

oleander spark jobs upload tutorial_part08_dataframe.py

Now confirm that the jobs are uploaded:

bash

oleander spark jobs list

You should see your uploaded entrypoints there.

To submit the SQL version:

bash

oleander spark jobs submit tutorial_part08_sql.py \
  --namespace tutorial \
  --name tutorial_part08_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_part08_dataframe.py \
  --namespace tutorial \
  --name tutorial_part08_dataframe \
  --args=--data-interval-start=2026-01-01T00:00:00Z \
  --args=--data-interval-end=2026-01-02T00:00:00Z

You might wonder what --namespace and --name are for.

Those are not Spark concepts. They are Oleander job identifiers.

In practice, you might use a team name or project name as the namespace. The job name gives you a distinct identity for one specific run configuration. That becomes useful when you run the same code in different contexts.

For example, you might run the same job every day on a regular schedule. Later, if you need to rerun it over a larger interval to backfill older data, you may want a different job name such as one with a _backfill_yyyymmdd suffix.

Summary

In this part, we took the raw clickstream table and started turning it into more useful summary tables.

We created three kinds of summaries:

daily traffic summaries
session summaries
session-product summaries

We wrote each one twice: once in Spark SQL and once in the DataFrame API. Along the way, we also introduced a few practical ideas that show up often in real Spark jobs, including parameterized date ranges, UTC handling, grouping and aggregation, lazy execution, and job submission.

From the next part on, we can start using these summary tables to do more meaningful analysis.

PreviousSetting Up a Retail Clickstream Table with Apache Iceberg NextAnalyzing Product Funnels with Aggregations