From e2bfbeb880e90d1c28423658c91b67e5b026242e Mon Sep 17 00:00:00 2001 From: muraliktalluri Date: Sat, 15 Aug 2026 16:27:57 -0500 Subject: [PATCH] Add RTM stream-stream join example: real-time ad click attribution --- .../README.md | 334 ++++++++++++++++++ .../RTM-StreamStreamJoin/RTM-SSJ.py | 288 +++++++++++++++ .../Write_RTM_attributed_clicks_to_delta.py | 75 ++++ .../debug.sql | 95 +++++ .../ingest-source-data/Kafka-clicks-ingest.py | 69 ++++ .../Kafka-impressions-ingest.py | 71 ++++ .../create-delete-topic-scala.scala | 65 ++++ .../generate-fake-adclick-data.py | 270 ++++++++++++++ 8 files changed, 1267 insertions(+) create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/README.md create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/RTM-SSJ.py create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/Write_RTM_attributed_clicks_to_delta.py create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/debug.sql create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-clicks-ingest.py create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-impressions-ingest.py create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/create-delete-topic-scala.scala create mode 100644 2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/generate-fake-adclick-data.py diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/README.md b/2026-08-rtm-streamStreamJoin-adClickAttribution/README.md new file mode 100644 index 0000000..a138348 --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/README.md @@ -0,0 +1,334 @@ +# Stream-Stream Joins in Apache Spark™ Real-Time Mode: Ad Click Attribution + +**Real-Time Mode (RTM)** in Structured Streaming now supports **stream-stream joins** (Databricks Runtime 18+). This is a **self-serve sample** you can import into **Databricks** and run end-to-end: a real-time **ad click attribution** pipeline built as an **inner, time-bounded stream-stream join** of two Kafka topics, running in Real-Time Mode with **sub-second end-to-end latency**. + +Joining two live streams is one of the most common — and most latency-sensitive — patterns in operational streaming. Until now, doing it with sub-second latency in Spark meant reaching for a second engine. RTM closes that gap: the **same Structured Streaming stream-stream join you already write** now runs in Real-Time Mode with a **single trigger change**. + +You bring **Kafka**, **Unity Catalog**, and a **Databricks Runtime** that supports this workload; we provide the **notebooks** and the **data generator / replay** path so a team can reproduce it in their own workspace. **RTM stream-stream join requires latest DBR 18.x** (18.2+ recommended). + +### Before you run: fill in the placeholders + +The notebooks use placeholders for workspace-specific values. Replace these across the repo (search-and-replace) with your own: + +| Placeholder | Replace with | +|-------------|--------------| +| `` | Your Unity Catalog catalog | +| `` | Your schema | +| `` | Databricks secret scope holding your Kafka bootstrap servers | +| `` | Secret key for the Kafka bootstrap servers | +| `` | Your cloud storage path for the checkpoint volume (e.g. `s3://…`) | + +### Companion blog post + +**TODO:** When the companion blog is published, paste the **full URL** below. + +**Blog post:** *`[add full https://… link when published]`*. + +--- + +## What's new: stream-stream joins in Real-Time Mode + +Real-Time Mode is a trigger type for Structured Streaming that delivers ultra-low (sub-second) end-to-end latency by executing all stages of a query concurrently and streaming data between them, rather than in discrete micro-batches. With DBR 18+, RTM adds support for **stateful stream-stream joins**, with a few characteristics to design around: + +- **Inner join only** (outer joins are not supported in RTM). +- **`update` output mode only.** +- **Both sides require watermarks**, and the join must include an **explicit time bound** so state stays bounded. +- A few **Spark configurations** enable it (shown below), plus the standard cluster-level RTM requirements (classic compute, no autoscaling, no Photon, DBR 18+). + +This repo shows exactly how to build such a join for a real, latency-sensitive use case — and how switching an existing micro-batch join to RTM is a one-line change. + +--- + +## The use case: connecting clicks back to impressions + +In digital advertising, two things happen in two different systems: + +- The **ad server** renders an ad → emits an **impression** event (rich context: campaign, advertiser, publisher, bid price, device, geo). +- The **click tracker** records a click → emits a **click** event (intentionally thin: `click_id`, `impression_id`, `device_id`, `click_time`). + +These arrive as **two independent, continuous streams**. A click by itself is nearly useless — it only carries the `impression_id` and a timestamp. To make it actionable (billing, budget pacing, click-fraud detection, feeding CTR back to bidding models) you must **join each click back to its impression** on `impression_id`, **within a short time window**. + +That is exactly a **stream-stream join**: + +``` +impressions ── join on impression_id ──┐ + ├──▶ attributed_clicks +clicks ─────────────────────────────────┘ + (click within 2 minutes of its impression) +``` + +**Why this is a fit for RTM:** it is naturally an **inner join** (only impression+click pairs that actually match are billable), it needs **event-time watermarks + a time-bounded condition**, and **latency has direct business value** — real-time CPC billing, budget pacing (stale data overspends), and click-fraud blocking all degrade when attribution lags. These are the operational workloads RTM is built for. + +### Why it is challenging + +``` +Timeline (one impression): +──────────────────────────────────────────────▶ time + │ │ + ▼ ▼ + Impression (Kafka) Click (Kafka, 0–2 min later) +``` + +- You must **hold impression state** until its click can no longer arrive (the 2-minute window), then evict it so state stays bounded. +- Both streams need **watermarks**, and the join needs an explicit **time bound**, or state grows without limit. +- **Late** clicks (past the window) and **orphan** clicks (no matching impression — the fraud/mismatch case) must be **dropped** — which the inner + time-bound semantics do for free. + +--- + +## What you'll learn + +1. How to build an **inner, time-bounded stream-stream join** in Structured Streaming and run it in **Real-Time Mode** (the enabling Spark configs, `maxPartitions`, slot budget, and `RealTimeTrigger`). +2. How **watermarks and state eviction** work for a stream-stream join, and how the time bound keeps state bounded. +3. How to **generate two correlated event streams** into Delta (100M impressions + ~10M clicks with deliberate *matched / late / orphan* cases) and replay them into Kafka at a controlled rate. +4. How to **observe the pipeline** with `StreamingQueryListener` (RTM `latencies` JSON, state metrics) and SQL percentiles — including a **latency comparison** between micro-batch and Real-Time Mode on the exact same code. + +--- + +## Project structure + +``` +RTM-join/ +├── debug.sql # SQL: expected-outcome oracle + E2E latency percentiles +├── ingest-source-data/ +│ ├── generate-fake-adclick-data.py # UC/Delta generator + time-sliced replay -> *_stream tables +│ ├── Kafka-impressions-ingest.py # Delta impressions_stream -> Kafka topic ad_impressions +│ ├── Kafka-clicks-ingest.py # Delta clicks_stream -> Kafka topic ad_clicks +│ └── create-delete-topic-scala.scala # Topic admin (Kafka AdminClient) +└── RTM-StreamStreamJoin/ + ├── RTM-SSJ.py # Main: Kafka x2 -> inner time-bounded join -> Kafka (RTM/MBM widget) + └── Write_RTM_attributed_clicks_to_delta.py # Optional: attributed_clicks (Kafka) -> Delta for SQL latency +``` + +**Fast path:** + +1. `generate-fake-adclick-data.py` → Delta `impressions_stream` / `clicks_stream`. +2. `create-delete-topic-scala.scala` → create `ad_impressions` (8 partitions), `ad_clicks` (2), `attributed_clicks` (8). +3. `RTM-SSJ.py` on the join cluster — `mode` = **RTM** or **MBM** — start it **first** so it is already reading. +4. `Kafka-impressions-ingest.py` then `Kafka-clicks-ingest.py` on separate cluster(s) to replay into Kafka. +5. (Optional) `Write_RTM_attributed_clicks_to_delta.py` → land `attributed_clicks` into Delta, then run `debug.sql`. + +--- + +## Prerequisites + +### 1. Databricks workspace / compute + +RTM stream-stream join requires a **classic** cluster on **latest DBR 18.x** (18.2+ recommended so `update` output mode is supported for stream-stream joins in **both** modes) with: + +- **No autoscaling, no Photon, no spot instances.** +- Cluster-level Spark conf: `spark.databricks.streaming.realTimeMode.enabled true`. +- Enough cores for the RTM slot budget (see below). The reference run used a **24-core** classic cluster. + +Ingest notebooks (Delta → Kafka) can run on a smaller **recent DBR LTS** cluster. + +### 2. Apache Kafka + +Bootstrap servers reachable from the cluster. Topics (default names — change to match yours): + +| Topic | Partitions | Purpose | +|-------|-----------|---------| +| `ad_impressions` | 8 | Impression JSON events | +| `ad_clicks` | 2 | Click JSON events | +| `attributed_clicks` | 8 | Join **sink** (both RTM and MBM write here) | + +Run `create-delete-topic-scala.scala` (Maven: `org.apache.kafka:kafka-clients:3.5.1`) to create/delete them. The script **deletes then recreates** — use only where safe. + +### 3. Databricks secrets + +Kafka bootstrap servers are read via secrets. Update the scope/key to your own: + +| Item | Value used in code | +|------|--------------------| +| Secret scope | `` | +| Secret key (Kafka bootstrap) | `` | + +```bash +databricks secrets create-scope --scope +databricks secrets put --scope --key +# paste comma-separated host:port list +``` + +All notebooks call `dbutils.secrets.get("", "")`. + +### 4. Unity Catalog: catalog, schema, volume + +| Object | Default in this repo | +|--------|----------------------| +| Catalog | `` | +| Schema | `` | +| External volume (checkpoints) | `..write_to_kafka` | +| Volume mount path | `/Volumes///write_to_kafka` | + +Create the external volume (fix the `LOCATION` in the `CREATE EXTERNAL VOLUME` SQL for your cloud) and update `volume_path` in the notebooks. If you use different catalog/schema names, search-replace `` / `` across the repo. + +--- + +## Building the join + +The heart of the demo is `RTM-SSJ.py`: read two Kafka topics, apply watermarks, and run an **inner, time-bounded join**. + +**Watermarks & join condition:** + +```python +impressions.withWatermark("impression_time", "5 minutes") +clicks.withWatermark("click_time", "5 minutes") + +# inner join: +# impressions.impression_id = clicks.impression_id +# AND click_time >= impression_time +# AND click_time <= impression_time + interval 2 minutes +``` + +**Enable stream-stream join in Real-Time Mode** (session-level; cluster-level `spark.databricks.streaming.realTimeMode.enabled=true` is also required): + +```python +spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true") +spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true") +spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4") +spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true") +spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2") +``` + +**One `mode` widget flips micro-batch ↔ Real-Time Mode — and only these settings change:** + +| Setting | RTM | MBM | +|---------|-----|-----| +| Trigger | `trigger(realTime="5 minutes")` | `trigger(processingTime="0.5 seconds")` | +| Kafka `maxPartitions` | impressions=8, clicks=2 | not set (default) | +| `spark.sql.shuffle.partitions` | `14` | `24` | +| Output mode | `update` | `update` (DBR 18.2+) | + +**RTM slot budget:** total task slots must be ≥ sum of tasks across stages. Here: `8 (impressions) + 2 (clicks) + 14 (shuffle) = 24` → run on a **24-core** cluster. + +### Watermarks and state eviction + +The global watermark is `min(impression_watermark, click_watermark)` (both 5 minutes). With the 2-minute join bound, Spark evicts each side once no future partner can match it: + +- **Impressions** are kept until `impression_time < globalWatermark − 2 min` — an impression waits up to 2 minutes for a click that may still arrive. +- **Clicks** are evicted at `click_time < globalWatermark` — a click's impression is always in the past, so it matches immediately or never. + +This keeps state **bounded**: with data flowing continuously, the join state store grows during warm-up, then plateaus as eviction rate matches insertion rate. A 5-minute watermark aligns with the 5-minute RTM checkpoint interval and guarantees every legitimate in-window click matches before its impression is evicted (the reference run dropped **0** records by watermark). + +--- + +## Running the reference test + +Run steps **in this order** so the join query is already live when ingest ramps. + +### Step 1 — Generate data: `generate-fake-adclick-data.py` + +Builds `impressions` and `clicks` Delta tables, then slices them into **one-file-per-second** `impressions_stream` / `clicks_stream` tables (in event-time order) for controlled replay. + +Key knobs (top of notebook): + +| Knob | Default | Meaning | +|------|---------|---------| +| `WINDOW_SECONDS` | `3600` | Event-time span (1 hour). Lower (e.g. `600`) for a quick test. | +| `NUM_IMPRESSIONS` | `100_000_000` | ~100M → ~27.7K impressions/sec | +| `CTR` | `0.10` | Fraction of impressions clicked (inflated for the demo) | +| `ATTRIBUTION_WINDOW_SECS` | `120` | The join time-bound (2 minutes) | +| `LATE_CLICK_FRACTION` | `0.10` | Clicks that land **after** the window → should not match | +| `ORPHAN_CLICK_FRACTION` | `0.02` | Clicks with no matching impression → dropped by inner join | + +Sanity-check cells report impressions/minute and the expected `matched / late / orphan` breakdown — your **correctness oracle** for what the join should produce (~9M matched). + +### Step 2 — Create Kafka topics + +Run `create-delete-topic-scala.scala` (Maven `org.apache.kafka:kafka-clients:3.5.1`; detach/re-attach the notebook after the library installs). Creates `ad_impressions` (8), `ad_clicks` (2), `attributed_clicks` (8). + +### Step 3 — Start the join first: `RTM-SSJ.py` + +Kafka `ad_impressions` + `ad_clicks` → inner time-bounded join → Kafka `attributed_clicks`. Widgets: `mode` = `RTM` or `MBM`; `clean_checkpoint` = `yes` (recommended when switching modes). + +### Step 4 — Start ingest: `Kafka-impressions-ingest.py`, then `Kafka-clicks-ingest.py` + +Each reads its `_stream` Delta table with `maxFilesPerTrigger=1` at `processingTime="1 second"` → **1 second of event-time per wall-second**, preserving order. Start impressions first, clicks 1–2s later. Combined ~**30K events/sec** (~27.7K impressions + ~2.7K clicks). + +### Step 5 — (Optional) Land results in Delta + `debug.sql` + +`Write_RTM_attributed_clicks_to_delta.py` subscribes to `attributed_clicks`, captures the Kafka `output_timestamp`, and writes a Delta table (`attributed_clicks_rtm` / `attributed_clicks_mbm`). Then `debug.sql` computes end-to-end latency percentiles: `timestampdiff(MILLISECOND, click_kafka_timestamp, output_timestamp)`. + +--- + +## How it performs: micro-batch vs Real-Time Mode + +Because the pipeline runs the **same code** in both modes (only the trigger changes), it's a clean way to see what Real-Time Mode buys you. End-to-end latency = **click's Kafka timestamp → `attributed_clicks` Kafka timestamp**; same 24-core DBR 18.x cluster, same data (~30K events/sec, ~9M attributed clicks). + +| Percentile | MBM | RTM | MBM ÷ RTM | +|-----------|----:|----:|----------:| +| min | 556 ms | 3 ms | ~185× | +| p50 (median) | 1,384 ms | 57 ms | ~24× | +| p90 | 1,692 ms | 85 ms | ~20× | +| p99 | 2,750 ms | 167 ms | ~16× | +| max | 6,084 ms | 2,599 ms | ~2.3× | + +**MBM's ~1-second batches are its latency floor** — a matched record can't be emitted until its batch completes. **RTM flows records through continuously** (single-digit-ms min), so median drops ~24× and p99 ~16×, with **zero records dropped** and **bounded state**. Your cluster will differ — treat this as a baseline and reproduce with the listener / `debug.sql`. + +--- + +## Data model + +### Impression (Kafka `value` JSON) + +```json +{ + "impression_id": "imp_000000009402", + "ad_id": "ad_004217", + "campaign_id": "camp_0381", + "advertiser_id": "adv_0057", + "publisher_id": "pub_0123", + "placement_id": "plc_07", + "device_id": "dev_000482915", + "device_type": "mobile", + "os": "iOS", + "geo_country": "US", + "geo_city": "New York", + "bid_price_usd": 0.0124, + "impression_time": "2025-11-01T00:06:45.000Z" +} +``` + +### Click (Kafka `value` JSON) — thin + +```json +{ + "click_id": "clk_5b81e0a2c3d4e5f6", + "impression_id": "imp_000000009402", + "device_id": "dev_000482915", + "click_time": "2025-11-01T00:06:52.000Z" +} +``` + +### Attributed click (join output → `attributed_clicks`) + +The join glues the thin click onto its rich impression and adds `time_to_click_secs` plus both Kafka timestamps (`impression_kafka_timestamp`, `click_kafka_timestamp`) for latency analysis. + +--- + +## Architecture + +``` +┌──────────────────┐ ┌─────────────────────┐ ┌──────────────────────────┐ +│ Delta replay │──▶│ Delta → Kafka │──▶│ ad_impressions / │ +│ (*_stream tables)│ │ (ingest notebooks) │ │ ad_clicks topics │ +└──────────────────┘ └─────────────────────┘ └────────────┬─────────────┘ + │ + ▼ +┌──────────────────┐ ┌─────────────────────┐ ┌──────────────────────────┐ +│ Optional Delta │◀──│ attributed_clicks │◀──│ inner time-bounded │ +│ sink + debug.sql │ │ (Kafka) │ │ stream-stream join (RTM-SSJ)│ +└──────────────────┘ └─────────────────────┘ └──────────────────────────┘ +``` + +--- + +## Additional resources + +- [Real-time mode in Structured Streaming (Databricks)](https://docs.databricks.com/aws/en/structured-streaming/real-time/concepts) +- [Set up real-time mode](https://docs.databricks.com/aws/en/structured-streaming/real-time/setup) +- [Real-time mode reference](https://docs.databricks.com/aws/en/structured-streaming/real-time/reference) +- [Stream-stream joins on Databricks](https://docs.databricks.com/aws/en/transform/join) + +--- + +Happy streaming. diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/RTM-SSJ.py b/2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/RTM-SSJ.py new file mode 100644 index 0000000..c8aa16f --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/RTM-SSJ.py @@ -0,0 +1,288 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Real-Time Mode: Stream-Stream Join for Ad Click Attribution +# MAGIC +# MAGIC Joins two Kafka streams — `ad_impressions` (an ad was shown) and `ad_clicks` (an ad was clicked) — +# MAGIC on `impression_id`, attributing each click to the impression that caused it, within a **2-minute** +# MAGIC attribution window. The enriched (attributed) click is written to the `attributed_clicks` Kafka topic. +# MAGIC +# MAGIC A `mode` widget flips between **RTM** (Real-Time Mode) and **MBM** (micro-batch mode) on the same code +# MAGIC — only the trigger, Kafka `maxPartitions`, and shuffle partitions change. +# MAGIC +# MAGIC RTM stream-stream join constraints: **inner join only**, **update output mode only**, both sides need +# MAGIC watermarks + a time-bounded join condition (DBR 18+). + +# COMMAND ---------- + +from pyspark.sql import functions as F +from pyspark.sql.functions import col, expr, from_json, struct, to_json +from pyspark.sql.types import StructType, StringType, DoubleType, TimestampType +from pyspark.sql.streaming import StreamingQueryListener + +# COMMAND ---------- + +dbutils.widgets.dropdown("mode", "RTM", ["RTM", "MBM"]) +mode = dbutils.widgets.get("mode") +print(f"Running in mode: {mode}") + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Spark configuration +# MAGIC +# MAGIC RocksDB state store + the stream-stream join configs required for RTM. The RTM enablement flag +# MAGIC (`spark.databricks.streaming.realTimeMode.enabled`) must also be set at the **cluster** level. + +# COMMAND ---------- + +if mode == "RTM": + # Required configs to enable stream-stream joins in Real-Time Mode (DBR 18+) + spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true") + spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true") + # RTM Slot allocation: source tasks (impressions maxPartitions 8 + clicks maxPartitions 2) + shuffle tasks must be <= total cluster cores. + spark.conf.set("spark.sql.shuffle.partitions", "14") +else: + spark.conf.set("spark.sql.shuffle.partitions", "24") + # Async checkpointing is enabled by default for RTM, but can be disabled for MBM + spark.conf.set("spark.databricks.streaming.statefulOperator.asyncCheckpoint.enabled","true") + +spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4") +spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true") +spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2") + + +# COMMAND ---------- + +stream_name = "RTM-adclick-ssj" +volume_path = "/Volumes///write_to_kafka" +checkpoint_path = f"{volume_path}/{stream_name}" + +dbutils.widgets.text("clean_checkpoint", "yes") +clean_checkpoint = dbutils.widgets.get("clean_checkpoint") +if clean_checkpoint == "yes": + dbutils.fs.rm(checkpoint_path, True) + +# COMMAND ---------- + +import json + +class CustomStreamingQueryListener(StreamingQueryListener): + def onQueryStarted(self, event): + print(f"Query started: id={event.id}, name={event.name}") + + def onQueryProgress(self, event): + row = event.progress + print("****************************************** batchId ***********************") + print( + f"batchId = {row.batchId} " + f"timestamp = {row.timestamp} " + f"numInputRows = {row.numInputRows} " + f"batchDuration = {row.batchDuration}" + ) + # state operator metrics (join state size, etc.) + for i, so in enumerate(row.stateOperators): + print( + f"stateOperator[{i}] numRowsTotal = {so.numRowsTotal} " + f"numRowsUpdated = {so.numRowsUpdated} " + f"memoryUsedBytes = {so.memoryUsedBytes}" + ) + # RTM-only: end-to-end latency percentiles from the engine + if mode == "RTM": + progress_json = json.loads(row.json) + latencies = progress_json.get("latencies", {}) + print(json.dumps(latencies, indent=2)) + + def onQueryTerminated(self, event): + print(f"Query terminated: id={event.id}, runId={event.runId}") + +try: + spark.streams.removeListener(CustomStreamingQueryListener()) +except Exception: + pass +spark.streams.addListener(CustomStreamingQueryListener()) + +# COMMAND ---------- + +kafka_bootstrap_servers_plaintext = dbutils.secrets.get("", "") +impressions_topic = "ad_impressions" +clicks_topic = "ad_clicks" +output_topic = "attributed_clicks" + +# COMMAND ---------- + +# MAGIC %md ## Schemas for the two input streams + +# COMMAND ---------- + +impressions_schema = ( + StructType() + .add("impression_id", StringType()) + .add("ad_id", StringType()) + .add("campaign_id", StringType()) + .add("advertiser_id", StringType()) + .add("publisher_id", StringType()) + .add("placement_id", StringType()) + .add("device_id", StringType()) + .add("device_type", StringType()) + .add("os", StringType()) + .add("geo_country", StringType()) + .add("geo_city", StringType()) + .add("bid_price_usd", DoubleType()) + .add("impression_time", TimestampType()) +) + +clicks_schema = ( + StructType() + .add("click_id", StringType()) + .add("impression_id", StringType()) + .add("device_id", StringType()) + .add("click_time", TimestampType()) +) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Read impressions stream +# MAGIC +# MAGIC `maxPartitions=8` (RTM only) coalesces the 8 Kafka partitions into 8 source tasks. We keep the Kafka +# MAGIC log timestamp as `impression_kafka_timestamp` for later end-to-end latency analysis. + +# COMMAND ---------- + +impressions_reader = ( + spark.readStream + .format("kafka") + .option("kafka.bootstrap.servers", kafka_bootstrap_servers_plaintext) + .option("subscribe", impressions_topic) + .option("startingOffsets", "earliest") +) +if mode == "RTM": + impressions_reader = impressions_reader.option("maxPartitions", 8) + +impressions = ( + impressions_reader.load() + .withColumnRenamed("timestamp", "impression_kafka_timestamp") + .withColumn("imp", from_json(col("value").cast("string"), impressions_schema)) + .select( + col("imp.impression_id").alias("impression_id"), + col("imp.ad_id").alias("ad_id"), + col("imp.campaign_id").alias("campaign_id"), + col("imp.advertiser_id").alias("advertiser_id"), + col("imp.publisher_id").alias("publisher_id"), + col("imp.placement_id").alias("placement_id"), + col("imp.device_id").alias("imp_device_id"), + col("imp.device_type").alias("device_type"), + col("imp.os").alias("os"), + col("imp.geo_country").alias("geo_country"), + col("imp.geo_city").alias("geo_city"), + col("imp.bid_price_usd").alias("bid_price_usd"), + col("imp.impression_time").alias("impression_time"), + col("impression_kafka_timestamp"), + ) + # 5-minute watermark: aligned with the 5-minute RTM checkpoint interval + .withWatermark("impression_time", "5 minutes") +) + +# COMMAND ---------- + +# MAGIC %md ## Read clicks stream (2 Kafka partitions) + +# COMMAND ---------- + +clicks_reader = ( + spark.readStream + .format("kafka") + .option("kafka.bootstrap.servers", kafka_bootstrap_servers_plaintext) + .option("subscribe", clicks_topic) + .option("startingOffsets", "earliest") +) +if mode == "RTM": + clicks_reader = clicks_reader.option("maxPartitions", 2) + +clicks = ( + clicks_reader.load() + .withColumnRenamed("timestamp", "click_kafka_timestamp") + .withColumn("clk", from_json(col("value").cast("string"), clicks_schema)) + .select( + col("clk.click_id").alias("click_id"), + col("clk.impression_id").alias("impression_id"), + col("clk.device_id").alias("click_device_id"), + col("clk.click_time").alias("click_time"), + col("click_kafka_timestamp"), + ) + # 5-minute watermark on the click side as well + .withWatermark("click_time", "5 minutes") +) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Inner, time-bounded stream-stream join +# MAGIC +# MAGIC A click is attributed to an impression only if it shares the same `impression_id` AND happens within +# MAGIC **2 minutes** after the impression. The time bound + watermarks let Spark evict impressions from state +# MAGIC ~7 minutes after their event time (2 min window + 5 min watermark), keeping state bounded. + +# COMMAND ---------- + +attributed = ( + impressions.alias("impressions").join( + clicks.alias("clicks"), + expr( + """ + impressions.impression_id = clicks.impression_id AND + click_time >= impression_time AND + click_time <= impression_time + interval 2 minutes + """ + ), + "inner", + ) + .select( + col("click_id"), + col("impressions.impression_id").alias("impression_id"), + col("ad_id"), + col("campaign_id"), + col("advertiser_id"), + col("publisher_id"), + col("placement_id"), + col("imp_device_id").alias("device_id"), + col("device_type"), + col("os"), + col("geo_country"), + col("geo_city"), + col("bid_price_usd"), + col("impression_time"), + col("click_time"), + (col("click_time").cast("long") - col("impression_time").cast("long")).alias("time_to_click_secs"), + col("impression_kafka_timestamp"), + col("click_kafka_timestamp"), + ) +) + +# COMMAND ---------- + +# MAGIC %md ## Write attributed clicks to Kafka (`attributed_clicks`) + +# COMMAND ---------- + +output_df = attributed.select( + col("impression_id").cast("binary").alias("key"), + to_json(struct("*")).cast("binary").alias("value"), +) + +# COMMAND ---------- + +query = ( + output_df.writeStream + .queryName("adclick-attribution") + .format("kafka") + .option("kafka.bootstrap.servers", kafka_bootstrap_servers_plaintext) + .option("topic", output_topic) + .option("checkpointLocation", checkpoint_path) + .outputMode("update") + .trigger(**({"realTime": "5 minutes"} if mode == "RTM" else {"processingTime": "0.5 seconds"})) + .start() +) + +# COMMAND ---------- + diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/Write_RTM_attributed_clicks_to_delta.py b/2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/Write_RTM_attributed_clicks_to_delta.py new file mode 100644 index 0000000..a1cf8e2 --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/RTM-StreamStreamJoin/Write_RTM_attributed_clicks_to_delta.py @@ -0,0 +1,75 @@ +# Databricks notebook source +from pyspark.sql.functions import col, from_json +from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType, TimestampType + +# COMMAND ---------- + +kafka_bootstrap_servers_plaintext = dbutils.secrets.get("", "") +output_topic = "attributed_clicks" +volume_path = '/Volumes///write_to_kafka' +checkpoint_path = f'{volume_path}/{output_topic}_to_delta' + +dbutils.widgets.text('clean_checkpoint', 'yes') +clean_checkpoint = dbutils.widgets.get('clean_checkpoint') +if clean_checkpoint == 'yes': + dbutils.fs.rm(checkpoint_path, True) + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC drop table if exists ..attributed_clicks_mbm; + +# COMMAND ---------- + +# Schema matches the attributed-click JSON emitted by RTM-SSJ.py (struct("*") of the join output) +kafka_schema = StructType([ + StructField("click_id", StringType(), True), + StructField("impression_id", StringType(), True), + StructField("ad_id", StringType(), True), + StructField("campaign_id", StringType(), True), + StructField("advertiser_id", StringType(), True), + StructField("publisher_id", StringType(), True), + StructField("placement_id", StringType(), True), + StructField("device_id", StringType(), True), + StructField("device_type", StringType(), True), + StructField("os", StringType(), True), + StructField("geo_country", StringType(), True), + StructField("geo_city", StringType(), True), + StructField("bid_price_usd", DoubleType(), True), + StructField("impression_time", TimestampType(), True), + StructField("click_time", TimestampType(), True), + StructField("time_to_click_secs", LongType(), True), + StructField("impression_kafka_timestamp", TimestampType(), True), + StructField("click_kafka_timestamp", TimestampType(), True), +]) + +# COMMAND ---------- + +stream_df = ( + spark.readStream.format("kafka") + .option("kafka.bootstrap.servers", kafka_bootstrap_servers_plaintext) + .option("subscribe", output_topic) + .option("startingOffsets", "earliest") + .load() + .withColumn("value", col("value").cast('string')) + .withColumn("value_struct", from_json(col("value"), kafka_schema)) + .selectExpr( + 'timestamp as output_timestamp', + 'value_struct.*' + ) +) + +# COMMAND ---------- + +( + stream_df + .writeStream + .queryName('write_RTM_attributed_clicks') + .outputMode("append") + .trigger(processingTime='1 seconds') + .option("checkpointLocation", checkpoint_path) + .toTable("..attributed_clicks_mbm") +) + +# COMMAND ---------- + diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/debug.sql b/2026-08-rtm-streamStreamJoin-adClickAttribution/debug.sql new file mode 100644 index 0000000..6cf1ac6 --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/debug.sql @@ -0,0 +1,95 @@ +-- ============================================================================ +-- Ad Click Attribution (RTM stream-stream join) — debug / results SQL +-- Table landed by Write_RTM_attributed_clicks_to_delta.py: +-- ..attributed_clicks_rtm +-- ============================================================================ + + +-- ---------------------------------------------------------------------------- +-- Expected join outcome (correctness oracle) — batch left join on the source +-- tables. matched (<= 2 min) is the row count the stream-stream join should +-- produce in attributed_clicks. +-- ---------------------------------------------------------------------------- +with j as ( + select + c.click_id, + i.impression_id as matched_impression, + (unix_timestamp(c.click_time) - unix_timestamp(i.impression_time)) as delay_secs + from ..clicks c + left join ..impressions i + on c.impression_id = i.impression_id +) +select + case + when matched_impression is null then 'orphan (no impression)' + when delay_secs <= 120 then 'matched (<= 2 min)' + else 'late (> 2 min)' + end as outcome, + count(1) as clicks +from j group by 1 order by 2 desc; + + +-- ---------------------------------------------------------------------------- +-- Actual: number of attributed clicks produced (should ~= matched count above) +-- ---------------------------------------------------------------------------- +select count(1) as attributed_clicks +from ..attributed_clicks_rtm; + + +-- ---------------------------------------------------------------------------- +-- Inspect a sample attributed row +-- ---------------------------------------------------------------------------- +select * +from ..attributed_clicks_rtm +limit 20; + + +-- ============================================================================ +-- LATENCY — CLICK side (THE headline metric) +-- E2E = click lands on Kafka -> attributed row lands on output topic. +-- The attributed row is produced when the click arrives and matches, so this +-- is the true end-to-end processing latency of the join. +-- ============================================================================ +with tab1 as ( + select + timestampdiff(MILLISECOND, click_kafka_timestamp, output_timestamp) as latency_ms + from ..attributed_clicks_rtm +) +select + count(1) as cnt, + min(latency_ms) as min, + percentile(latency_ms, 0.10) as p10, + percentile(latency_ms, 0.25) as p25, + percentile(latency_ms, 0.50) as median, + percentile(latency_ms, 0.75) as p75, + percentile(latency_ms, 0.90) as p90, + percentile(latency_ms, 0.95) as p95, + percentile(latency_ms, 0.99) as p99, + max(latency_ms) as max +from tab1; + + +-- ============================================================================ +-- LATENCY — IMPRESSION side (context only, NOT the performance metric) +-- E2E = impression lands on Kafka -> attributed row lands on output topic. +-- This is inflated by the time the impression sits idle in state waiting for +-- its click (up to the 2-min window) — mostly business time-to-click, not +-- engine latency. Useful only to contrast against the click-side number. +-- ============================================================================ +with tab1 as ( + select + timestampdiff(MILLISECOND, impression_kafka_timestamp, output_timestamp) as latency_ms + from ..attributed_clicks_rtm +) +select + count(1) as cnt, + min(latency_ms) as min, + percentile(latency_ms, 0.10) as p10, + percentile(latency_ms, 0.25) as p25, + percentile(latency_ms, 0.50) as median, + percentile(latency_ms, 0.75) as p75, + percentile(latency_ms, 0.90) as p90, + percentile(latency_ms, 0.95) as p95, + percentile(latency_ms, 0.99) as p99, + max(latency_ms) as max +from tab1; diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-clicks-ingest.py b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-clicks-ingest.py new file mode 100644 index 0000000..8dd973a --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-clicks-ingest.py @@ -0,0 +1,69 @@ +# Databricks notebook source +from pyspark.sql import functions as F +from pyspark.sql.streaming import StreamingQueryListener + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC CREATE EXTERNAL VOLUME IF NOT EXISTS ..write_to_kafka +# MAGIC LOCATION 's3:///ad_click_write_to_kafka/' + +# COMMAND ---------- + +kafka_bootstrap_servers_plaintext = dbutils.secrets.get("", "") +clicks_topic = 'ad_clicks' +volume_path = '/Volumes///write_to_kafka' +checkpoint_path = f'{volume_path}/{clicks_topic}' + +dbutils.widgets.text('clean_checkpoint', 'yes') +clean_checkpoint = dbutils.widgets.get('clean_checkpoint') +if clean_checkpoint == 'yes': + dbutils.fs.rm(checkpoint_path, True) + +# COMMAND ---------- + +class MyStreamingListener(StreamingQueryListener): + def onQueryStarted(self, event): + print(f"'{event.name}' [{event.id}] got started!") + def onQueryProgress(self, event): + row = event.progress + print(f"****************************************** batchId ***********************") + print(f"batchId = {row.batchId} timestamp = {row.timestamp} numInputRows = {row.numInputRows} batchDuration = {row.batchDuration}") + def onQueryTerminated(self, event): + print(f"{event.id} got terminated!") + +try: + spark.streams.removeListener(MyStreamingListener()) +except: + pass +spark.streams.addListener(MyStreamingListener()) + +# COMMAND ---------- + +clicks_stream_df = ( + spark.readStream + .format("delta") + .option("maxFilesPerTrigger", 1) # 1 file per trigger = 1 second's worth of data + .table("..clicks_stream") + .withColumn("all_columns", F.to_json(F.struct( + 'click_id', 'impression_id', 'device_id', 'click_time' + ))) + .selectExpr('CAST(impression_id AS BINARY) AS key', 'CAST(all_columns AS BINARY) AS value') +) + +# COMMAND ---------- + +( + clicks_stream_df + .writeStream + .queryName('ad_clicks_ingest') + .format('kafka') + .option("kafka.bootstrap.servers", kafka_bootstrap_servers_plaintext) + .option("topic", clicks_topic) + .option("checkpointLocation", checkpoint_path) + .trigger(processingTime = '1 seconds') + .start() +) + +# COMMAND ---------- + diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-impressions-ingest.py b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-impressions-ingest.py new file mode 100644 index 0000000..6f03c68 --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/Kafka-impressions-ingest.py @@ -0,0 +1,71 @@ +# Databricks notebook source +from pyspark.sql import functions as F +from pyspark.sql.streaming import StreamingQueryListener + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC CREATE EXTERNAL VOLUME IF NOT EXISTS ..write_to_kafka +# MAGIC LOCATION 's3:///ad_click_write_to_kafka/' + +# COMMAND ---------- + +kafka_bootstrap_servers_plaintext = dbutils.secrets.get("", "") +impressions_topic = 'ad_impressions' +volume_path = '/Volumes///write_to_kafka' +checkpoint_path = f'{volume_path}/{impressions_topic}' + +dbutils.widgets.text('clean_checkpoint', 'yes') +clean_checkpoint = dbutils.widgets.get('clean_checkpoint') +if clean_checkpoint == 'yes': + dbutils.fs.rm(checkpoint_path, True) + +# COMMAND ---------- + +class MyStreamingListener(StreamingQueryListener): + def onQueryStarted(self, event): + print(f"'{event.name}' [{event.id}] got started!") + def onQueryProgress(self, event): + row = event.progress + print(f"****************************************** batchId ***********************") + print(f"batchId = {row.batchId} timestamp = {row.timestamp} numInputRows = {row.numInputRows} batchDuration = {row.batchDuration}") + def onQueryTerminated(self, event): + print(f"{event.id} got terminated!") + +try: + spark.streams.removeListener(MyStreamingListener()) +except: + pass +spark.streams.addListener(MyStreamingListener()) + +# COMMAND ---------- + +impressions_stream_df = ( + spark.readStream + .format("delta") + .option("maxFilesPerTrigger", 1) # 1 file per trigger = 1 second's worth of data from the generator + .table("..impressions_stream") + .withColumn("all_columns", F.to_json(F.struct( + 'impression_id', 'ad_id', 'campaign_id', 'advertiser_id', + 'publisher_id', 'placement_id', 'device_id', 'device_type', + 'os', 'geo_country', 'geo_city', 'bid_price_usd', 'impression_time' + ))) + .selectExpr('CAST(impression_id AS BINARY) AS key', 'CAST(all_columns AS BINARY) AS value') +) + +# COMMAND ---------- + +( + impressions_stream_df + .writeStream + .queryName('ad_impressions_ingest') + .format('kafka') + .option("kafka.bootstrap.servers", kafka_bootstrap_servers_plaintext) + .option("topic", impressions_topic) + .option("checkpointLocation", checkpoint_path) + .trigger(processingTime = '1 seconds') + .start() +) + +# COMMAND ---------- + diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/create-delete-topic-scala.scala b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/create-delete-topic-scala.scala new file mode 100644 index 0000000..0755dde --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/create-delete-topic-scala.scala @@ -0,0 +1,65 @@ +// Databricks notebook source +// Databricks notebook source +// Maven dependency: org.apache.kafka:kafka-clients:3.5.1 + +// COMMAND ---------- + +import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig, NewTopic} +import java.util.{Collections, Properties} +import scala.jdk.CollectionConverters._ + +// COMMAND ---------- + +val kafkaBootstrapServers = dbutils.secrets.get("", "") + +val impressionsTopic = "ad_impressions" +val clicksTopic = "ad_clicks" +val outputTopic = "attributed_clicks" + +// COMMAND ---------- + +val props = new Properties() +props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers) +val adminClient = AdminClient.create(props) + +// COMMAND ---------- + +def deleteTopic(topicName: String): Unit = { + try { + adminClient.deleteTopics(Collections.singletonList(topicName)).all().get() + println(s"Topic '$topicName' deleted.") + } catch { + case e: Exception => println(s"Failed to delete topic '$topicName': ${e.getMessage}") + } +} + +deleteTopic(impressionsTopic) +deleteTopic(clicksTopic) +deleteTopic(outputTopic) + +// COMMAND ---------- + +val retentionMs = -1 + +def createTopic(topicName: String, numPartitions: Int, replicationFactor: Short = 3): Unit = { + val topic = new NewTopic(topicName, numPartitions, replicationFactor) + .configs(Map("retention.ms" -> retentionMs.toString).asJava) + try { + adminClient.createTopics(Collections.singletonList(topic)).all().get() + println(s"Topic '$topicName' created successfully.") + } catch { + case e: Exception => println(s"Error creating topic '$topicName': ${e.getMessage}") + } +} + +// 8 partitions for impressions, 2 for clicks, 8 for output +createTopic(impressionsTopic, 8) +createTopic(clicksTopic, 2) +createTopic(outputTopic, 8) + +// COMMAND ---------- + +adminClient.close() + +// COMMAND ---------- + diff --git a/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/generate-fake-adclick-data.py b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/generate-fake-adclick-data.py new file mode 100644 index 0000000..9a53603 --- /dev/null +++ b/2026-08-rtm-streamStreamJoin-adClickAttribution/ingest-source-data/generate-fake-adclick-data.py @@ -0,0 +1,270 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Generate fake ad-click data (impressions + clicks) for the RTM stream-stream join blog +# MAGIC +# MAGIC Produces two correlated event streams for a real-time **ad click attribution** demo: +# MAGIC - `impressions` — an ad was shown (rich context: campaign, advertiser, publisher, bid price, device, geo) +# MAGIC - `clicks` — an ad was clicked (thin: click_id + impression_id + device_id + click_time) +# MAGIC +# MAGIC The pipeline then joins `clicks` back to `impressions` on `impression_id` within a time window. +# MAGIC +# MAGIC Scale target: ~100M impressions over a 1-hour window (3600 one-second buckets => ~27.7K impressions/sec), +# MAGIC ~10% CTR => ~10M clicks. Data is generated set-based in Spark (not on the driver) so it scales to 100M. +# MAGIC The `_stream` tables are sliced into 1-second buckets (one Delta file per second) and replayed 1 file/sec +# MAGIC by the Kafka ingest notebooks, preserving event-time order. + +# COMMAND ---------- + +from pyspark.sql import functions as F +from pyspark.sql.functions import col, lit, array, element_at + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC use catalog ; +# MAGIC create schema if not exists ; +# MAGIC use schema ; +# MAGIC +# MAGIC ALTER SCHEMA DISABLE PREDICTIVE OPTIMIZATION; + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC -- keep one file per write (repartition(1)) so the 1-second replay slices map to single files +# MAGIC set spark.databricks.delta.autoCompact.enabled = false; +# MAGIC set spark.databricks.delta.optimizeWrite.enabled = false; + +# COMMAND ---------- + +CATALOG = "" +SCHEMA = "" + +# --- Volume / window knobs --------------------------------------------------- +BASE_TIME = "2025-11-01T00:00:00" +WINDOW_SECONDS = 3600 # 1 hour. Lower this (e.g. 600) for a fast functional test. +NUM_IMPRESSIONS = 100_000_000 # ~100M => ~27.7K impressions/sec across the window +CTR = 0.10 # fraction of impressions that get clicked (inflated for the demo) +ROWS_PER_SECOND = NUM_IMPRESSIONS // WINDOW_SECONDS # ~27.7K rows per 1-second bucket + +# --- Attribution / click-timing knobs -------------------------------------- +ATTRIBUTION_WINDOW_SECS = 120 # 2 min: the join time-bound (click within impression_time + 2 min) +LATE_CLICK_FRACTION = 0.10 # of clicked impressions, ~10% click AFTER the window (should NOT match) +ORPHAN_CLICK_FRACTION = 0.02 # extra clicks with no matching impression (fraud/mismatch -> dropped by inner join) + +# --- Dimension cardinalities (hierarchy: ad -> campaign -> advertiser) ------- +NUM_ADVERTISERS = 500 +NUM_CAMPAIGNS = 5_000 +NUM_ADS = 50_000 +NUM_PUBLISHERS = 1_000 +NUM_PLACEMENTS = 20 +NUM_DEVICES = 20_000_000 + +ADS_PER_CAMPAIGN = NUM_ADS // NUM_CAMPAIGNS # 10 +CAMPAIGNS_PER_ADVERTISER = NUM_CAMPAIGNS // NUM_ADVERTISERS # 10 + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC drop table if exists ..impressions; +# MAGIC drop table if exists ..clicks; + +# COMMAND ---------- + +# Shared base epoch (session-tz consistent) for all timestamp math +base_col = F.unix_timestamp(F.lit(BASE_TIME).cast("timestamp")) + +geo_city = array(*[lit(c) for c in [ + "New York", "Los Angeles", "Chicago", "Houston", "Seattle", + "London", "Toronto", "Sydney", "Berlin", "Mumbai", +]]) +geo_country = array(*[lit(c) for c in [ + "US", "US", "US", "US", "US", + "UK", "CA", "AU", "DE", "IN", +]]) + +# COMMAND ---------- + +# Generate impressions set-based. Contiguous second buckets (id / ROWS_PER_SECOND) spread rows evenly +# across the hour (each of the 3600 buckets gets ~ROWS_PER_SECOND rows). +impressions_df = ( + spark.range(NUM_IMPRESSIONS) + .withColumn("second_bucket", F.least((col("id") / ROWS_PER_SECOND).cast("int"), lit(WINDOW_SECONDS - 1))) + .withColumn("impression_time", (base_col + col("second_bucket") + F.rand()).cast("timestamp")) + .withColumn("impression_id", F.format_string("imp_%012d", col("id"))) + # ad -> campaign -> advertiser, kept internally consistent via integer arithmetic + .withColumn("ad_num", (F.rand() * NUM_ADS).cast("long")) + .withColumn("campaign_num", (col("ad_num") / ADS_PER_CAMPAIGN).cast("long")) + .withColumn("advertiser_num", (col("campaign_num") / CAMPAIGNS_PER_ADVERTISER).cast("long")) + .withColumn("ad_id", F.format_string("ad_%06d", col("ad_num"))) + .withColumn("campaign_id", F.format_string("camp_%04d", col("campaign_num"))) + .withColumn("advertiser_id", F.format_string("adv_%04d", col("advertiser_num"))) + .withColumn("publisher_id", F.format_string("pub_%04d", (F.rand() * NUM_PUBLISHERS).cast("long"))) + .withColumn("placement_id", F.format_string("plc_%02d", (F.rand() * NUM_PLACEMENTS).cast("long"))) + .withColumn("device_id", F.format_string("dev_%09d", (F.rand() * NUM_DEVICES).cast("long"))) + .withColumn("device_type", element_at(array(lit("mobile"), lit("desktop"), lit("ctv")), (F.rand() * 3 + 1).cast("int"))) + .withColumn( + "os", + F.when(col("device_type") == "mobile", element_at(array(lit("iOS"), lit("Android")), (F.rand() * 2 + 1).cast("int"))) + .when(col("device_type") == "desktop", element_at(array(lit("Windows"), lit("macOS")), (F.rand() * 2 + 1).cast("int"))) + .otherwise(element_at(array(lit("Roku"), lit("FireTV"), lit("AndroidTV")), (F.rand() * 3 + 1).cast("int"))), + ) + .withColumn("geo_idx", (F.rand() * 10 + 1).cast("int")) + .withColumn("geo_city", element_at(geo_city, col("geo_idx"))) + .withColumn("geo_country", element_at(geo_country, col("geo_idx"))) + .withColumn("bid_price_usd", F.round(F.rand() * 0.009 + 0.001, 4)) # $0.001 - $0.010 per impression + .select( + "impression_id", "ad_id", "campaign_id", "advertiser_id", + "publisher_id", "placement_id", "device_id", "device_type", "os", + "geo_country", "geo_city", "bid_price_usd", "impression_time", + "second_bucket", + ) +) + +( + impressions_df.write.format("delta").mode("overwrite") + .saveAsTable(f"{CATALOG}.{SCHEMA}.impressions") +) +print(f"Wrote impressions table (~{NUM_IMPRESSIONS:,} rows)") + +# COMMAND ---------- + +# Derive clicks FROM impressions so the join key actually correlates. +# ~90% of clicked impressions click inside the window (matches), ~10% after it (late -> dropped by the join). +impressions_tbl = spark.table(f"{CATALOG}.{SCHEMA}.impressions") + +clicks_from_impr = ( + impressions_tbl + .sample(withReplacement=False, fraction=CTR, seed=42) + .withColumn("_late_roll", F.rand()) + .withColumn( + "click_delay_secs", + F.when( + col("_late_roll") >= LATE_CLICK_FRACTION, + # matched: right-skewed within [0, ATTRIBUTION_WINDOW) -> most clicks happen early + F.floor(F.pow(F.rand(), lit(2.0)) * ATTRIBUTION_WINDOW_SECS), + ).otherwise( + # late: [ATTRIBUTION_WINDOW, 2*ATTRIBUTION_WINDOW) -> past the bound, should not match + ATTRIBUTION_WINDOW_SECS + F.floor(F.rand() * ATTRIBUTION_WINDOW_SECS), + ), + ) + .withColumn("click_time", (col("impression_time").cast("double") + col("click_delay_secs") + F.rand()).cast("timestamp")) + .withColumn("click_id", F.concat(lit("clk_"), F.substring(F.md5(col("impression_id")), 1, 16))) + .select("click_id", "impression_id", "device_id", "click_time") +) + +# COMMAND ---------- + +# Orphan clicks: impression_ids beyond the real range => never match (fraud / mismatched clicks). +expected_clicks = int(NUM_IMPRESSIONS * CTR) +num_orphans = int(expected_clicks * ORPHAN_CLICK_FRACTION) + +orphan_clicks = ( + spark.range(num_orphans) + .withColumn("impression_id", F.format_string("imp_%012d", col("id") + NUM_IMPRESSIONS)) + .withColumn("device_id", F.format_string("dev_%09d", (F.rand() * NUM_DEVICES).cast("long"))) + .withColumn("click_time", (base_col + F.rand() * WINDOW_SECONDS).cast("timestamp")) + .withColumn("click_id", F.concat(lit("clk_orphan_"), F.format_string("%012d", col("id")))) + .select("click_id", "impression_id", "device_id", "click_time") +) + +clicks_df = ( + clicks_from_impr.unionByName(orphan_clicks) + .withColumn("click_second_bucket", F.floor(col("click_time").cast("double") - base_col).cast("int")) + .filter(col("click_second_bucket") >= 0) +) + +( + clicks_df.write.format("delta").mode("overwrite") + .saveAsTable(f"{CATALOG}.{SCHEMA}.clicks") +) +print(f"Wrote clicks table (~{expected_clicks:,} matched/late + {num_orphans:,} orphans) partitioned by click_second_bucket") + +# COMMAND ---------- + +# MAGIC %md ## Sanity checks + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC -- impressions per minute (should be ~ NUM_IMPRESSIONS/60 each) +# MAGIC select date_trunc('MINUTE', impression_time) as minute, count(1) as impressions +# MAGIC from ..impressions +# MAGIC group by 1 order by 1; + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC -- expected join outcome breakdown: matched (in window) vs late vs orphan +# MAGIC with j as ( +# MAGIC select +# MAGIC c.click_id, +# MAGIC i.impression_id as matched_impression, +# MAGIC (unix_timestamp(c.click_time) - unix_timestamp(i.impression_time)) as delay_secs +# MAGIC from ..clicks c +# MAGIC left join ..impressions i +# MAGIC on c.impression_id = i.impression_id +# MAGIC ) +# MAGIC select +# MAGIC case +# MAGIC when matched_impression is null then 'orphan (no impression)' +# MAGIC when delay_secs <= 120 then 'matched (<= 2 min)' +# MAGIC else 'late (> 2 min)' +# MAGIC end as outcome, +# MAGIC count(1) as clicks +# MAGIC from j group by 1 order by 2 desc; + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Build the replay `_stream` tables (1 file per second, in event-time order) +# MAGIC +# MAGIC Each second-bucket is written as its own Delta commit (`repartition(1)` => one file), in order, +# MAGIC so the Kafka ingest notebooks can replay them with `maxFilesPerTrigger=1` at 1 file/sec. +# MAGIC +# MAGIC NOTE: this is a one-time prep step of `WINDOW_SECONDS` sequential writes and takes a while at 1h scale. +# MAGIC For a quick functional test, set `WINDOW_SECONDS` small (e.g. 600) at the top and re-run. + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC drop table if exists ..impressions_stream; +# MAGIC drop table if exists ..clicks_stream; + +# COMMAND ---------- + +# impressions_stream: one file per second-bucket, in order +impressions_tbl = spark.table(f"{CATALOG}.{SCHEMA}.impressions") +for s in range(WINDOW_SECONDS): + ( + impressions_tbl.filter(col("second_bucket") == s) + .drop("second_bucket") + .repartition(1) + .write.format("delta").mode("append") + .saveAsTable(f"{CATALOG}.{SCHEMA}.impressions_stream") + ) + if s % 300 == 0: + print(f"impressions_stream: wrote second {s}/{WINDOW_SECONDS}") +print("impressions_stream complete") + +# COMMAND ---------- + +# clicks_stream: one file per click-second-bucket, in order (clicks can spill past the hour by up to the window) +clicks_tbl = spark.table(f"{CATALOG}.{SCHEMA}.clicks") +max_click_bucket = clicks_tbl.agg(F.max("click_second_bucket")).collect()[0][0] +for s in range(int(max_click_bucket) + 1): + ( + clicks_tbl.filter(col("click_second_bucket") == s) + .drop("click_second_bucket") + .repartition(1) + .write.format("delta").mode("append") + .saveAsTable(f"{CATALOG}.{SCHEMA}.clicks_stream") + ) + if s % 300 == 0: + print(f"clicks_stream: wrote second {s}/{int(max_click_bucket) + 1}") +print("clicks_stream complete") + +# COMMAND ---------- + +# MAGIC %sql +# MAGIC ALTER TABLE ..impressions_stream DISABLE PREDICTIVE OPTIMIZATION; +# MAGIC ALTER TABLE ..clicks_stream DISABLE PREDICTIVE OPTIMIZATION; \ No newline at end of file