Background
It is 05:30 AM in Copenhagen, Denmark. I am the Lead Data Engineer for NordicStream, the largest Real-Time Bidding (RTB) ad exchange in Scandinavia.
The platform processes 400,000 ad auctions per second, connecting publishers (like news sites) with advertisers. Every night, a critical batch pipeline runs to join the raw Bid Logs (5 TB) with the User Dimensions (500 GB) to calculate the "Revenue Per User Segment."
The Stakes: This data powers the "Daily Spend Dashboard" used by major advertisers (Lego, Maersk, Carlsberg). If the dashboard is empty at 08:00 AM, they pause their campaigns.
The Cost: Every hour of downtime costs the company €150,000 in lost bidding fees.
The Business Crisis
-
The setting: A Real - Time Bidding (RTB) platform processing 400,000 auctions per second.
-
The Problem: A critical batch pipeline that usually takes 45 minutes has been running for 6 hours and is stuck at 99%.
Technical Bottlenecks
-
Data Skew: 1.2 billion records have an "unknown" device_id, causing a single Spark partition to process 100GB while others remain idle.
-
Inefficient Code: The use of explode() on nested JSON arrays is creating billions of intermediate rows, leading to "Shuffle Spills" to disk.
-
Python GIL Lock: Legacy Python loops for parsing User Agents are slowing processing by 10x.
-
Data Quality: Kafka "At-Least-Once" delivery has introduced 5% duplicate records that must be resolved to ensure exact revenue reporting.
Deliverables
Phase 1: Optimization (Solving the skew):
-
Deliverable 1 (Concept): A markdown cell explaining why a Broadcast Join cannot be used (based on the 500GB table size) and why the standard shuffle join failed.
-
Deliverable 2 (Code): PySpark code to implement a Salted Join, involving adding a random salt (0-19) to the Bid Logs, replicating the User Dimension rows (via array explosion) to match the salt range and joining the tables on both device_id and salt.
Phase 2: Efficient Transformation
-
Deliverable 3 (Vectorization): A Pandas UDF named parse_agent_udf that extracts the Operating System (iOS/Android) from the user_agent string, including proof of correct return values.
-
Deliverable 4 (Higher-Order Functions): Rewriting the aggregation logic using transform or aggregate functions to sum bid_price within the impressions array without using explode().
Phase 3: Storage Architecture
-
Deliverable 5 (Bucketing): Writing the final User Dimensions table using bucketBy on the device_id column with 50 buckets.
Phase 4: Advanced Analytics (Window Functions)
-
Deliverable 6 (Deduplication Pattern): Implementing a "Rank & Filter" pattern to handle Kafka duplicates, define a Window Specification partitioned by bid_id and ordered by timestamp descending AND use row_number() to generate a rank and filter for rank == 1 to keep the latest record.
-
Deliverable 7 (Running Total Pattern): Calculating "Cumulative Spend" for the Lego campaign, define a Window Specification partitioned by campaign_id and ordered by timestamp ascending AND use sum(bid_price) over the window to create the cumulative_spend column.
Workspace
Actionables
The below code snippet is the initial setup for ingestion in our Databricks notebook, designed to load the raw "Bid Logs" into a PySpark DataFrame. It begins by importing all necessary Spark SQL types, such as StructType and TimestampType, to define a formal schema named bids_schema. However, in the subsequent spark.read.csv command, this manual schema is not actually applied; instead, the code uses inferSchema=True, which forces Spark to scan the entire CSV file to guess the data types, a process that can be slow and inaccurate for large datasets. Finally, the display(df_bids) function is called to render a scrollable, visual preview of the data directly within the notebook interface.
We are now loading the Users Dimensions table, which is one of the two primary datasets required for our join operations. The spark.read.csv function reads the file from Databricks Volume, using header=True to treat the first row as column names and inferSchema=True to automatically detect data types like strings or integers. Finally, display(df_users) provides a visual grid of the user data so we can verify the contents before beginning the transformation phases.
Then we are initiating the Salted Join technique, which is a key requirement for Phase 1 of our case study. This code addresses the Black Friday Data Skew where 1.2 billion records share the same "unknown" ID, causing a single Spark partition to bottleneck the entire pipeline. By using F.rand() multiplied by a SALT_FACTOR of 20, we are appending a new random integer column called salt to wer massive Bid Logs table. This splits that massive "unknown" key into 20 different sub-keys, allowing Spark to distribute the workload across 20 different executors instead of crushing just one.
In this fourth cell, we are preparing the User Dimensions (the "Right" table) to match the salted Bid Logs. Since we added a random salt to the bid records, we must ensure every user record exists for every possible salt value (0-19) so the join doesn't miss any data. This code creates a salt_array containing 20 integers for every single user row and then uses F.explode() to flatten that array, effectively replicating the entire dimension table 20 times.
In this fifth cell, we are executing the final Inner Join between our salted Bid Logs and the replicated User Dimensions. By joining on both device_id and the newly created salt column, we have successfully redistributed the massive "unknown" device workload across 20 different Spark partitions. This directly resolves the "Black Friday" Skew described in the case study, where a single task was previously stuck for 4.5 hours processing 100GB of data while the rest of the cluster sat idle.
In this sixth cell, we are fulfilling Deliverable 3 by implementing a Pandas UDF (User Defined Function) to extract the Operating System from the user_agent string. This approach is a direct response to the "Legacy Code" warning in the case study, which noted that standard Python loops were causing a "Python GIL" lock and slowing processing by 10x. By using a Pandas UDF, Spark utilizes Apache Arrow to transfer data in batches (vectorization) between the JVM and Python, bypassing the high overhead of row-by-row serialization found in traditional Python UDFs.
In this seventh cell, we are fulfilling Deliverable 6 by implementing the "Rank & Filter" pattern to solve the Deduplication Problem. Because the Kafka pipeline uses "At-Least-Once" delivery, some bid logs are duplicates that must be cleaned to ensure exact revenue reporting. This code uses a Window Function to group records by bid_id and sort them by timestamp in descending order. By applying row_number() and filtering for the first rank, we discard older duplicates and keep only the most recent version of every bid.
In this eighth cell, we are fulfilling Deliverable 7 by calculating a Running Total of advertising spend. This addresses the requirement for advertisers to monitor their hourly spending limits in real-time. By defining a Window Specification partitioned by campaign_id and ordered by timestamp in ascending order, the F.sum() function calculates the cumulative sum of total_bid_cost for each row relative to the previous ones in that specific campaign.
