Streaming Clickstream Events into Iceberg

In Part 7, we downloaded a pre-populated Parquet file and inserted it into retail_clickstream_events.

That worked well for learning Iceberg and Spark. It gave us a useful source table without first building an entire data-ingestion system, which would have been a rather ambitious warm-up exercise.

In a real retail website, however, clickstream data does not arrive as one convenient daily file.

When a customer opens a page, searches for a product, adds something to a cart, or completes a purchase, that individual action generates an event. Events continue to happen while customers use the website.

The Spark jobs we have written so far process a bounded chunk of data for a requested date interval. This is batch processing.

For the raw clickstream table, we want a long-running application that receives new events and writes them as they happen. This is streaming ingestion.

Spark Structured Streaming gives us a way to build that application.

It is not the only option. We could use the Iceberg sink for Kafka Connect, or write a streaming application with Apache Flink. This is an Iceberg and Spark tutorial, though, so let’s make the predictable choice and use Spark.

What We Are Going to Build

Our ingestion flow will look like this:

Oleander retail event
        ↓
Kafka topic
        ↓
Spark Structured Streaming application
        ↓
oleander.tutorial.retail_clickstream_events

Kafka will hold the incoming events. Spark will continually read those events, parse and validate their JSON values, and append valid rows to the existing Iceberg table.

Although we call this streaming, Spark will not create one Iceberg commit for every individual event. Structured Streaming groups available events into small batches called micro-batches.

Our application uses a one-minute trigger. Once per minute, Spark checks for available events and processes the next micro-batch. A test event may therefore take roughly one minute to appear in Iceberg.

This is still streaming ingestion. “Streaming” does not necessarily mean that every row must arrive before we have time to blink.

Running Kafka and Spark Locally

To make this hands-on exercise self-contained, we will run both sides of the streaming connection on your computer.

Kafka will run in Docker and listen on localhost. The Spark streaming application will run as a separate local process with spark-submit.sh. The application will read from local Kafka and write through the configured Oleander catalog to our Iceberg table.

There is a practical reason for this arrangement.

If we submitted the streaming application to Oleander-managed Spark, localhost would refer to the remote Spark environment—not your computer. The remote job would have no way to reach the Kafka broker running on your laptop.

This does not mean a streaming application must always run locally. In a real deployment, you need a proper Kafka cluster whose network address is reachable from the Spark environment. With that setup, and the appropriate Kafka security configuration, you can run the streaming application with Oleander-managed Spark streaming instead.

We will therefore run Spark locally with the generated spark-submit.sh wrapper.

The Spark driver and Kafka broker will both run on your computer. The wrapper will still configure Spark with your Oleander catalog, so the local application can write to the same Iceberg table used throughout this series.

Our Kafka setup uses one broker with plaintext networking. It is intentionally small and convenient for local learning. A production Kafka cluster would usually have multiple brokers, authentication, encryption, monitoring, and people who become concerned when someone says, “It is only one little event.”

Preparing the Workspace

Initialize the Part 15 workspace:

oleander spark init part15-streaming

Use:

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

Then enter the workspace and install its local dependencies:

cd part15-streaming
uv sync

Part 4 explains the generated project and local wrappers in more detail. In this part, we will use both spark-submit.sh and pyspark.sh.

Running Kafka with Docker Compose

Create a file named docker-compose.yml in the Part 15 workspace and put the following configuration in it:

services:
  kafka:
    image: apache/kafka:3.9.2
    hostname: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: CONTROLLER://:29093,PLAINTEXT_HOST://:9092,PLAINTEXT://:19092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT_HOST://localhost:9092,PLAINTEXT://kafka:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      CLUSTER_ID: 4L6g3nShT-eMCtK--X86sw
    healthcheck:
      test:
        - CMD-SHELL
        - /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1
      interval: 5s
      timeout: 5s
      retries: 12

  kafka-init:
    image: apache/kafka:3.9.2
    depends_on:
      kafka:
        condition: service_healthy
    command:
      - /opt/kafka/bin/kafka-topics.sh
      - --bootstrap-server
      - kafka:19092
      - --create
      - --if-not-exists
      - --topic
      - retail-clickstream-events
      - --partitions
      - "1"
      - --replication-factor
      - "1"

This is based on Apache Kafka’s official single-node plaintext Docker example.

The kafka service runs one Kafka process as both a broker and controller. This mode is enough for a local exercise; we do not need to build a miniature data center before lunch.

The configuration has two addresses that matter to us:

• programs on your computer connect through localhost:9092
• other Compose services connect through kafka:19092

The health check waits until Kafka can answer a command. After that, kafka-init creates a one-partition topic named retail-clickstream-events. The --if-not-exists option makes topic creation safe if the initializer runs again against the same broker.

The replication settings are one because there is only one broker.

Start the services:

docker compose up -d

Check their status:

docker compose ps -a

The kafka service should be healthy. The kafka-init service should exit with status code 0 after creating the topic.

We can also inspect the topic directly:

docker compose exec kafka \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic retail-clickstream-events

The result should report one partition and a replication factor of one.

Configuring the Streaming Application

The application receives three pieces of configuration.

Two come from environment variables:

• KAFKA_BOOTSTRAP contains the Kafka bootstrap address
• KAFKA_TOPIC contains the topic name

The third comes from Spark configuration:

• spark.oleander.app.state.dir contains a directory where the application can keep its state

These names are defined near the top of the entrypoint:

KAFKA_BOOTSTRAP_ENV = "KAFKA_BOOTSTRAP"
KAFKA_TOPIC_ENV = "KAFKA_TOPIC"
APP_STATE_DIR_CONF = "spark.oleander.app.state.dir"
CHECKPOINT_DIRECTORY = "part15-retail-clickstream-events"
TARGET_TABLE = "oleander.tutorial.retail_clickstream_events"

The environment variables are required:

def get_required_environment_variable(name: str) -> str:
    value = os.environ.get(name)
    if value is None or not value.strip():
        raise ValueError(f"Required environment variable {name} is not set")
    return value.strip()

The Spark configuration is used to construct a stable checkpoint location:

def get_checkpoint_location(spark: SparkSession) -> str:
    state_directory = spark.conf.get(APP_STATE_DIR_CONF, "").strip()
    if not state_directory:
        raise ValueError(
            f"Required Spark configuration {APP_STATE_DIR_CONF} is not set"
        )

    state_directory = state_directory.rstrip("/")
    return f"{state_directory}/{CHECKPOINT_DIRECTORY}"

The application fails immediately if any required value is missing. A streaming application cannot do much useful work if it does not know where its stream begins, where its table lives, or where to remember its progress.

The target table must already exist. This application transforms and writes events; it does not recreate retail_clickstream_events.

Defining the JSON Schema

Kafka messages contain a key and a value as bytes. Our producer will put one JSON object in each message value.

Spark needs to know how the JSON fields map to Spark types, so we define the schema explicitly:

CLICKSTREAM_EVENT_SCHEMA = StructType(
    [
        StructField("event_id", StringType(), True),
        StructField("event_time", TimestampType(), True),
        StructField("user_id", StringType(), True),
        StructField("session_id", StringType(), True),
        StructField("event_type", StringType(), True),
        StructField("page_url", StringType(), True),
        StructField("referrer_url", StringType(), True),
        StructField("search_query", StringType(), True),
        StructField("product_id", StringType(), True),
        StructField("product_name", StringType(), True),
        StructField("category", StringType(), True),
        StructField("cart_id", StringType(), True),
        StructField("order_id", StringType(), True),
        StructField("quantity", IntegerType(), True),
        StructField("price", DecimalType(10, 2), True),
        StructField("device_type", StringType(), True),
        StructField("browser", StringType(), True),
        StructField("traffic_source", StringType(), True),
    ]
)

This schema has 18 columns. The destination table has 19.

The missing column is ingestion_time. A producer knows when the customer action happened, so it supplies event_time. The Spark application knows when it processed the event, so Spark assigns ingestion_time itself.

The application also defines the destination order explicitly:

OUTPUT_COLUMNS = (
    "event_id",
    "event_time",
    "ingestion_time",
    "user_id",
    "session_id",
    "event_type",
    "page_url",
    "referrer_url",
    "search_query",
    "product_id",
    "product_name",
    "category",
    "cart_id",
    "order_id",
    "quantity",
    "price",
    "device_type",
    "browser",
    "traffic_source",
)

The explicit schema also tells Spark to parse event_time as a timestamp, quantity as an integer, and price as DECIMAL(10,2). JSON fields not listed in this schema are ignored.

All fields are nullable at the parsing stage. This lets parsing and business validation remain separate steps. Nullable does not mean every null is acceptable; it means we will produce a more useful validation error than “something somewhere was null.”

Reading Events from Kafka

Here is the Kafka source:

def read_kafka_events(
    spark: SparkSession,
    kafka_bootstrap: str,
    kafka_topic: str,
) -> DataFrame:
    return (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", kafka_bootstrap)
        .option("subscribe", kafka_topic)
        .option("startingOffsets", "earliest")
        .load()
    )

readStream is the streaming counterpart of the batch read interface we used earlier.

The kafka format tells Spark to use its Kafka source. We provide the broker address and subscribe to one topic.

startingOffsets controls where a brand-new stream begins. earliest tells Spark to read from the earliest message still available in Kafka, including messages published before the application started.

That setting applies only when the checkpoint is new. After the application has recorded progress, the checkpointed offsets decide where a restart resumes.

The returned DataFrame is a streaming DataFrame. It describes an input that can continue receiving rows rather than a fixed collection of rows that already exists.

Parsing and Validating Events

First, we cast Kafka’s binary value to a string and parse it as JSON:

parsed_events = (
    kafka_events.select(
        F.from_json(
            F.col("value").cast("string"),
            CLICKSTREAM_EVENT_SCHEMA,
            {"mode": "FAILFAST"},
        ).alias("event")
    )
    .select("event.*")
    .withColumn("_validation_error", build_validation_error())
)

FAILFAST makes malformed JSON or a value incompatible with the schema fail the micro-batch. For example, the string "several" cannot be parsed as an integer quantity.

Parsing successfully is not the same as describing a valid retail event, so the application also applies business rules.

Every event requires these fields:

• event_id
• event_time
• user_id
• session_id
• event_type
• page_url
• device_type
• browser
• traffic_source

Required string fields cannot be blank.

The allowed event types are page_view, search, product_view, add_to_cart, remove_from_cart, checkout_start, and purchase.

Some event types require additional information:

referrer_url remains optional, and the application does not require fields unrelated to an event type to be null.

The validation builder produces the first readable error for an invalid row. The application then makes that error part of the output expression:

validated_events = (
    parsed_events.withColumn(
        "event_id",
        F.when(
            F.col("_validation_error").isNotNull(),
            F.raise_error(
                F.concat(
                    F.lit("Invalid clickstream event: "),
                    F.col("_validation_error"),
                )
            ),
        ).otherwise(F.col("event_id")),
    )
    .drop("_validation_error")
    .withColumn("ingestion_time", F.current_timestamp())
)

raise_error() is evaluated when the streaming sink processes the row. An invalid event therefore fails its micro-batch instead of quietly disappearing.

After validation, current_timestamp() records the ingestion time. The application finishes by selecting all 19 columns in the same order as the destination table:

return validated_events.select(*OUTPUT_COLUMNS)

This application deliberately fails on invalid data. We will discuss the operational consequence later, but we will not poison our own tutorial stream just to prove that poison works.

Writing the Stream to Iceberg

Once the events have the correct schema, we can start the Iceberg sink:

def start_clickstream_ingestion(
    events: DataFrame,
    checkpoint_location: str,
) -> StreamingQuery:
    return (
        events.writeStream.format("iceberg")
        .outputMode("append")
        .trigger(processingTime="1 minute")
        .option("checkpointLocation", checkpoint_location)
        .option("fanout-enabled", "true")
        .queryName("part15-retail-clickstream-events")
        .toTable(TARGET_TABLE)
    )

writeStream is the streaming counterpart of the batch write interfaces used in earlier parts.

The Iceberg sink uses append mode because every micro-batch contains new clickstream events. toTable() writes them to the existing table.

The processing-time trigger asks Spark to start a micro-batch once per minute. Iceberg recommends avoiding very frequent commits because each commit creates table metadata and may create small data files. We will meet those small files properly in Part 16; there is no need to invite them all in at once.

In Part 13, we partitioned retail_clickstream_events by day(event_time). The fanout writer lets a streaming task write rows for multiple partition values without first performing an expensive global sort. Iceberg documents both the trigger and fanout considerations in its Structured Streaming guide.

queryName() gives the running stream a readable name. It does not create a table or Kafka topic.

The checkpoint location is what lets Spark remember which Kafka offsets and micro-batches it has already processed. It is part of the application’s correctness, not a temporary download directory.

The sink does not deduplicate event_id. If a producer sends the same valid event twice, this application can append it twice.

Keeping the Application Running

The main function connects the pieces:

def main() -> None:
    kafka_bootstrap = get_required_environment_variable(KAFKA_BOOTSTRAP_ENV)
    kafka_topic = get_required_environment_variable(KAFKA_TOPIC_ENV)

    spark = (
        SparkSession.builder.appName("tutorial-part15-streaming").getOrCreate()
    )
    streaming = None

    try:
        spark.conf.set("spark.sql.session.timeZone", "UTC")
        checkpoint_location = get_checkpoint_location(spark)
        kafka_events = read_kafka_events(spark, kafka_bootstrap, kafka_topic)
        clickstream_events = parse_and_validate_events(kafka_events)
        streaming = start_clickstream_ingestion(
            clickstream_events,
            checkpoint_location,
        )
        streaming.awaitTermination()
    finally:
        if streaming is not None and streaming.isActive:
            streaming.stop()
        spark.stop()

As before, the Spark session uses UTC so event timestamps and daily partitions have one consistent interpretation.

Unlike our batch jobs, this application has no date-range arguments and no natural end. awaitTermination() keeps the driver alive while the stream runs.

When the application terminates, the finally block stops an active stream and then stops Spark.

Starting the Streaming Application

Run the application from the Part 15 workspace:

KAFKA_BOOTSTRAP=localhost:9092 \
KAFKA_TOPIC=retail-clickstream-events \
./spark-submit.sh \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.5 \
  --conf "spark.oleander.app.state.dir=$PWD/.state" \
  tutorial_part15_streaming.py

The generated wrapper already supplies the Iceberg dependencies and Oleander catalog configuration. Spark’s Kafka connector is a separate package, so we add the connector matching Spark 3.5.5 with --packages.

The first run may download that package before starting the application.

We use an absolute .state path as the application state directory. The code adds part15-retail-clickstream-events beneath it for this stream’s checkpoint.

The command remains running. That is expected. A streaming application that exits immediately would be admirably efficient at ingesting nothing.

Leave this terminal open.

Publishing Clickstream Events

Open a second terminal in the Part 15 workspace and start Kafka’s console producer:

docker compose exec kafka \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic retail-clickstream-events

The producer waits for input. Paste the following JSON as one line and press Enter:

{
  "event_id": "evt_stream_000001",
  "event_time": "2026-01-09T12:00:00Z",
  "user_id": "user_101",
  "session_id": "session_stream_001",
  "event_type": "page_view",
  "page_url": "https://retail.oleander.dev/",
  "device_type": "desktop",
  "browser": "chrome",
  "traffic_source": "direct"
}

The streaming application checks Kafka once per one-minute trigger. Events that arrive before the next trigger can be processed together in the same micro-batch; an event does not normally require its own one-minute wait.

For this exercise, however, wait until the streaming application has processed the message before publishing the next one. Waiting a little more than one minute is the simplest approach. We are deliberately creating several small micro-batches—and therefore several small Iceberg files—so that Part 16 has files to compact.

Then paste the following events one at a time. Press Enter after each JSON line and wait for the next micro-batch to finish before moving to the next one:

{
  "event_id": "evt_stream_000002",
  "event_time": "2026-01-09T10:05:00Z",
  "user_id": "usr_stream_002",
  "session_id": "sess_stream_002",
  "event_type": "page_view",
  "page_url": "/collections/drinkware",
  "device_type": "desktop",
  "browser": "Chrome",
  "traffic_source": "direct"
}

{
  "event_id": "evt_stream_000003",
  "event_time": "2026-01-09T10:06:00Z",
  "user_id": "usr_stream_003",
  "session_id": "sess_stream_003",
  "event_type": "search",
  "page_url": "/search",
  "search_query": "trail mug",
  "device_type": "mobile",
  "browser": "Safari",
  "traffic_source": "organic_search"
}

{
  "event_id": "evt_stream_000004",
  "event_time": "2026-01-09T10:07:00Z",
  "user_id": "usr_stream_004",
  "session_id": "sess_stream_004",
  "event_type": "product_view",
  "page_url": "/products/trail-mug",
  "product_id": "P100",
  "product_name": "Trail Mug",
  "category": "drinkware",
  "price": 18.0,
  "device_type": "desktop",
  "browser": "Firefox",
  "traffic_source": "email"
}

{
  "event_id": "evt_stream_000005",
  "event_time": "2026-01-09T10:08:00Z",
  "user_id": "usr_stream_005",
  "session_id": "sess_stream_005",
  "event_type": "add_to_cart",
  "page_url": "/products/trail-mug",
  "product_id": "P100",
  "product_name": "Trail Mug",
  "category": "drinkware",
  "cart_id": "cart_stream_005",
  "quantity": 1,
  "price": 18.0,
  "device_type": "mobile",
  "browser": "Chrome",
  "traffic_source": "paid_search"
}

After publishing the fifth event, press Control-D to exit the producer.

One input line becomes one Kafka message. The JSON contains event_time, but not ingestion_time; Spark will add the latter when it processes the micro-batch.

All five events use January 9 so that they follow the January 8 data appended in Part 14. They also belong to the same day(event_time) partition.

Publishing them in separate micro-batches intentionally creates several small data files. That is not normally a goal worth celebrating, but it gives us something useful to compact in Part 16.

Verifying the Iceberg Rows

Allow up to roughly one minute for the trigger to process the final event.

Then open a third terminal in the Part 15 workspace and start the interactive shell:

./pyspark.sh

Set the Spark session timezone to UTC:

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

This is a newly started Spark session, so its first query resolves the current Iceberg table metadata from the catalog. There is no previously loaded table state to refresh.

Query the events by their IDs:

spark.sql("""
    SELECT
      event_id,
      event_type,
      event_time,
      ingestion_time,
      user_id,
      session_id,
      page_url
    FROM oleander.tutorial.retail_clickstream_events
    WHERE event_id IN (
      'evt_stream_000001',
      'evt_stream_000002',
      'evt_stream_000003',
      'evt_stream_000004',
      'evt_stream_000005'
    )
    ORDER BY event_id
""").show(truncate=False)

You should see five rows. Each event_time is the timestamp from its JSON message. The ingestion_time values depend on when your Spark application processed each event, so your values will naturally differ from everyone else’s.

If all five rows are not visible yet, wait for the next trigger and rerun the query.

Checkpoints and Restarts

The checkpoint records the stream’s progress, including the Kafka offsets already processed. The event data itself lives in Kafka and Iceberg; the checkpoint is the connection between those two histories.

If you stop the Spark application with Control-C and start it again with the same Kafka broker and checkpoint directory, Spark resumes from its recorded offsets. It does not return to the beginning merely because startingOffsets says earliest.

Deleting the checkpoint changes that behavior. With no saved offsets, the application treats itself as a new stream and reads from the earliest messages retained in Kafka. Because our Iceberg sink appends, replaying an already written message can create a duplicate row.

The Kafka setup in this article is ephemeral. It has no named data volume. Stopping and starting the existing container preserves its current data, but removing and recreating the Compose stack creates a fresh broker.

Do not keep an old Spark checkpoint after replacing Kafka with a fresh broker. The checkpoint refers to offsets from the previous broker, not the new one.

For a complete local reset, stop the Spark application and remove both sides of the local state:

docker compose down -v
rm -rf .state/part15-retail-clickstream-events

This resets Kafka and the Spark checkpoint. It does not remove rows already written to Iceberg. If you repeat the exercise afterward, use a new event_id or deliberately remove your earlier test row first.

In normal operation, checkpoints should be stored durably and preserved across application restarts. Our local directory is suitable for this exercise, not a production deployment design.

What Happens to an Invalid Event?

Malformed JSON, incompatible types, missing required fields, and invalid business values cause the current micro-batch to fail.

Spark does not advance the checkpoint for a failed micro-batch. The invalid Kafka message therefore remains in the uncommitted input range, and an unchanged restart encounters it again.

This is deliberate fail-fast behavior. It prevents bad data from silently entering Iceberg, but it also means someone must resolve the poison message before the stream can continue.

Production systems often put an ingestion API in front of Kafka. The website sends an event to the API, the API validates it, and only valid events are published to the main Kafka topic.

The API can reject malformed requests immediately or send them to a dead-letter topic for later inspection. This keeps many poison messages from reaching the Spark application in the first place.

Validation in the streaming application is still valuable as another line of defense. Producer bugs, schema changes, or events from another source can otherwise put invalid data into the topic despite the API boundary.

Dead-letter handling and controlled recovery are outside this beginner tutorial, so we will leave our valid example pleasantly unpoisoned.

Full Code

The complete companion entrypoint is available here:

• Part 15 streaming application

Unlike the earlier batch jobs, we do not build and upload this application to Oleander. It must run locally so it can reach the local Kafka broker.

Going Further

If you would like to see this idea in a broader production context, take a look at the Data Lineage in Production presentation.

The presentation also covers open table formats, compute engines, data lineage, and operational debugging. Starting from slide 15, it walks through a Kafka and Spark Structured Streaming example, including micro-batches, checkpoints, and the surrounding stream architecture.

Summary

In this part, we replaced our batch-file setup with a long-running Spark Structured Streaming application.

Kafka received individual Oleander retail events. readStream created a streaming DataFrame from the Kafka topic. Spark parsed each JSON value with an explicit schema, validated its business fields, and assigned ingestion_time.

writeStream processed events in one-minute micro-batches and appended them to the existing Iceberg table. The checkpoint recorded progress so a normal restart could resume from the correct Kafka offsets.

We also saw that streaming adds operational concerns beyond the transformation itself. Checkpoints must remain consistent with their source, invalid events need a recovery plan, and duplicate producer messages are not automatically removed.

Streaming writes also tend to create many small data files over time.

In the final part, we will inspect that small-file problem and use Iceberg compaction to combine small files into larger ones.