https://community.databricks.com/t5/technical-blog/apache-spark-s-real-time-mode-use-case-focus-ad-attribution/ba-p/146141 <H1 class="c22">Apache Spark's Real-time Mode Use Case Deep Dive: Ad Attribution</H1> <P class="c24" data-unlink="true"><SPAN>This article is a companion to <A href="https://www.databricks.com/blog/why-apache-spark-real-time-mode-game-changer-ad-attribution" target="_self">this&nbsp;</A></SPAN><A href="https://www.databricks.com/blog/why-apache-spark-real-time-mode-game-changer-ad-attribution" target="_self"><SPAN>Databricks blog</SPAN></A>&nbsp;<SPAN>&nbsp;about an Ad Attribution use case, with a&nbsp;</SPAN><SPAN>GitHub repo</SPAN><SUP><A id="cmnt_ref2" href="#cmnt2" target="_blank" rel="noopener">[b]</A></SUP><SPAN>&nbsp;including sample code that you can use to see</SPAN><SPAN class="c48"><A class="c19" href="https://www.google.com/url?q=https://docs.databricks.com/aws/en/structured-streaming/real-time&amp;sa=D&amp;source=editors&amp;ust=1769790659780565&amp;usg=AOvVaw3Kagv-BWZf60HNaGBvVTG0" target="_blank" rel="noopener">&nbsp;Real-Time Mode (RTM)</A></SPAN><SPAN class="c5 c21">&nbsp;in action.</SPAN></P> <P class="c24"><SPAN>In this article, we go deep into the underlying architecture and mechanics of real-time mode (RTM) while reflecting on the Ad Attribution use case and the example code. First, we define "operational" use cases and explain how Apache Spark</SPAN><SPAN class="c59">™</SPAN><SPAN>&nbsp;can now handle a new set of latency-sensitive use cases. Second, we review the architectural changes in Spark that enable RTM to achieve such low latency. Lastly, we explain some&nbsp;</SPAN><SPAN class="c5 c21"><SPAN>sample code, showing how, when coupled with&nbsp;<A href="https://www.google.com/url?q=https://docs.databricks.com/aws/en/stateful-applications/&amp;sa=D&amp;source=editors&amp;ust=1769790659781542&amp;usg=AOvVaw2M6kZHpLSJK5ir2BPkrtqI" target="_blank" rel="noopener">transformWithState</A>, ad attribution&nbsp;</SPAN>can be structured and plugged into a real-time stream. The sample code lets you compare the latency of RTM with that of the micro-batch execution mode.</SPAN></P> <H2 class="c42"><SPAN class="c39 c53 c21">Operational Workloads and Sub-Second Latency</SPAN></H2> <H3 class="c12"><SPAN class="c23">Apache Spark™ and Operational Use Cases</SPAN></H3> <P class="c24"><SPAN>Apache Spark Structured Streaming has long served as the industry standard for analytical streaming workloads, excelling at data ingestion and transformation, following architectures such as the Medallion pattern. However, a significant gap historically existed when addressing truly operational workloads—those systems requiring immediate response to data to trigger downstream actions or decisions, such as real-time fraud flagging or dynamic promotional messaging. These operational systems are characterized by a stringent requirement for sub-second, and often single-digit millisecond, end-to-end latency. Achieving this level of performance historically necessitated the use of specialized streaming engines outside the standard Spark/Databricks environment. This fragmentation compelled organizations to manage disparate data architectures, resulting in increased total cost of ownership (TCO) due to duplicated infrastructure, increased deployment complexity, resource contention, and the need to maintain multiple, distinct skill sets. The introduction of Apache Spark’s real-time mode, available in Databricks Runtime 16.4 LTS and above, directly addresses this issue. RTM delivers continuous, ultra-low-latency processing, pushing p99 latencies down to the single-digit milliseconds, effectively unifying the platform for both analytical and highly demanding operational streaming.</SPAN></P> <H3 class="c12"><SPAN class="c23">Architectural Trade-Offs: Micro-Batch vs. Continuous Execution vs. Real-Time Mode</SPAN></H3> <P class="c24"><SPAN class="c5 c21">The evolution of Structured Streaming execution models has been driven by the perpetual trade-off between latency, throughput, and stability.</SPAN></P> <H4 class="c49"><SPAN class="c56 c58">Micro-Batch Execution (V1)</SPAN></H4> <P class="c15"><SPAN class="c4">Micro-batching relies on the MicroBatchExecution engine, accumulating data into small, discrete batches.</SPAN><SPAN class="c35">&nbsp;</SPAN><SPAN class="c4">While this approach guarantees robust exactly-once processing and high throughput, its core limitation lies in the sequential nature of its execution loop. The execution phase cycles through distinct steps: fetching offsets, retrieving the batch data, planning the query, adding the batch, and finally, committing the write-ahead log (WAL). This sequential process, governed by a set trigger interval, introduces fixed scheduling latency that typically confines micro-batch performance to the seconds-latency range, making it unsuitable for true sub-second operational requirements.</SPAN></P> <H4 class="c32"><SPAN>Legacy Continuous Execution</SPAN></H4> <P class="c15"><SPAN class="c5 c4">Prior to RTM, Spark offered a "Continuous" execution mode, which processed records individually for lower latency. This experimental mode and didn't support a rich feature set, such as stateful queries.</SPAN></P> <H4 class="c49"><SPAN class="c56 c58">Real-Time Mode (RTM)</SPAN></H4> <P class="c15"><SPAN class="c4">RTM represents a fundamental architectural evolution that resolves the stability-vs-speed dilemma. RTM runs batches for longer durations (the default length is 5 minutes).<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c5 c4">Checkpoints, in this context, function similarly to batches.</SPAN><SPAN class="c4">&nbsp;</SPAN><SPAN class="c4">Critically, within this long batch window, RTM processes data<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c4 c62">as soon as it becomes available</SPAN><SPAN class="c5 c4">&nbsp;using concurrent scheduling. This design maintains the reliable, periodic checkpointing structure of a long-running batch while completely eliminating the scheduling coordination overhead of micro-batch mode. This hybridization strategy enables RTM to offer the robust fault tolerance associated with micro-batching, combined with sub-second latency performance.</SPAN></P> <H2 class="c42"><SPAN class="c39 c21 c53">Important: Cluster Configuration and Sizing</SPAN></H2> <P class="c15"><SPAN class="c5 c4">Successfully deploying RTM requires adhering to a precise set of cluster configurations and strict sizing methodologies. These parameters are not merely optimizations; they are the architectural keys that switch Spark’s behavior from micro-batching to continuous, concurrent processing.</SPAN></P> <H3 class="c12"><SPAN>Required Spark Configurations: Analyzing the Setup</SPAN></H3> <P class="c32"><SPAN class="c5 c4">The Ad Attribution workload demo provides the definitive configuration template necessary for RTM activation and high-performance operation:</SPAN></P> <UL class="c41 lst-kix_ik0d9knih2jt-0 start"> <LI class="c32 c43 li-bullet-0"><SPAN class="c28 c4">Activating RTM:</SPAN><SPAN class="c4 c5">&nbsp;Setting spark.databricks.streaming.realTimeMode.enabled true is the fundamental flag that enables the streaming query engine to support RTM.</SPAN></LI> </UL> <P class="c63"><SPAN class="c4">In Databricks, you can set the Spark confs in the Advanced section of the<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c48"><A class="c19" href="https://www.google.com/url?q=https://docs.databricks.com/aws/en/compute/configure&amp;sa=D&amp;source=editors&amp;ust=1769790659789016&amp;usg=AOvVaw2oNokaqaVWq3Lu_WvM7fQ7" target="_blank" rel="noopener">cluster configuration</A></SPAN><SPAN class="c4">.</SPAN><SPAN>&nbsp;</SPAN><SPAN class="c5 c21">Please note that this Spark conf is required for cluster bootup, unlike some other Spark confs that are session-scoped and changeable within runtime code.</SPAN></P> <H3 class="c12"><SPAN class="c23">The Task Slot Constraint: Ensuring Capacity Lockstep</SPAN></H3> <P class="c15"><SPAN class="c4">The most critical operational constraint in RTM sizing is the task slot requirement. To run effectively in ultra-low-latency mode, the total number of available task slots in the compute cluster<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c4 c28">must be greater than or equal to the total number of tasks across all query stages</SPAN><SPAN class="c4">.</SPAN></P> <P class="c15"><SPAN class="c4">This constraint is vital because the concurrent execution model requires every task in the query graph to have immediate access to dedicated compute resources. If the required number of slots is unavailable—for instance, if the workload generates 100 tasks but the cluster only provides 80 slots—an exception will be thrown that tells you the cluster does not have enough available slots.</SPAN></P> <P class="c15"><SPAN class="c4">This configuration necessitates a shift from reactive resource scaling (typical of batch ETL) to a proactive capacity reservation model. The cluster must be sized to meet the parallelism requirement of the peak load. Unlike micro-batch mode, where resources can be temporarily idled or shared, RTM tasks maintain their allocation within the long batch window, meaning precise right-sizing is crucial to prevent both resource exhaustion and unnecessary compute waste.</SPAN></P> <H3 class="c12"><SPAN class="c23">Tuning Data Parallelism: Optimizing Utilization</SPAN></H3> <P class="c32"><SPAN class="c5 c4">Since capacity is fixed and preemptively reserved, tuning the job’s parallelism is essential for optimizing resource utilization:</SPAN></P> <UL class="c41 lst-kix_me6s2c2s59nk-0 start"> <LI class="c32 c47 li-bullet-0"><SPAN class="c28 c4">Source Parallelism (maxPartitions):</SPAN><SPAN class="c5 c4">&nbsp;For input systems like Kafka, which feeds the Ad Attribution pipeline, Databricks recommends setting maxPartitions, such that each task handles multiple Kafka partitions. This reduces task overhead and balances data flow across the allocated compute slots. maxPartitions is a new option in Databricks Kafka connection options that was introduced along with RTM.</SPAN></LI> <LI class="c1 li-bullet-0"><SPAN class="c28 c4">Shuffle Parallelism:</SPAN><SPAN class="c4">&nbsp;For shuffle-heavy jobs, such as those involving stateful operations (transformWithState) or complex stream-stream joins, the number of shuffle partitions must be carefully chosen. Engineers should experiment to find the minimum number of shuffle partitions that prevent data backlogs. This count directly impacts the total required task slots (Source tasks + Shuffle tasks).</SPAN></LI> <LI class="c1 li-bullet-0"><SPAN class="c28 c4">Target Utilization:</SPAN><SPAN class="c5 c4">&nbsp;Tuning parallelism and overall cluster size often targets a goal utilization level (e.g., 50%). This buffer capacity ensures that the system can handle bursts of incoming data without violating the low-latency objective.</SPAN></LI> </UL> <P class="c14"><SPAN class="c5 c4">The physical capacity (task slots) and the logical parallelism (source and shuffle partitions) are inherently linked. If an increase in logical parallelism occurs without a corresponding increase in physical capacity, the RTM job cannot meet its concurrency requirement, resulting in performance degradation.</SPAN></P> <H3 class="c12"><SPAN class="c23">Production Best Practices</SPAN></H3> <P class="c32"><SPAN class="c5 c4">For scheduling RTM workloads, several operational best practices must be observed:</SPAN></P> <OL class="c41 lst-kix_gl8vof19ysde-0 start" start="1"> <LI class="c32 c43 li-bullet-0"><SPAN class="c28 c4">Compute Type:</SPAN><SPAN class="c4">&nbsp;Always schedule RTM streams as jobs using dedicated<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c28 c4">Jobs Compute</SPAN><SPAN class="c5 c4">. All-purpose compute clusters are not recommended for production streaming workloads.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">Scheduling Mode:</SPAN><SPAN class="c5 c4">&nbsp;Jobs should be scheduled using the conceptual "Continuous mode," which configures the job to run indefinitely and automatically restart upon failure.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">Autoscaling Prohibition:</SPAN><SPAN class="c4">&nbsp;Due to the rigid task slot constraint required by RTM,<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c28 c4">autoscaling must be disabled</SPAN><SPAN class="c4">.</SPAN><SPAN class="c35">&nbsp;</SPAN><SPAN class="c5 c4">Scaling delays would destroy the low-latency guarantees.</SPAN></LI> </OL> <H2 class="c42"><SPAN class="c39 c53 c21">Advanced Stateful Logic: Implementing Ad Attribution with transformWithState</SPAN></H2> <P class="c15"><SPAN class="c5 c4">RTM provides a high-performance execution environment, but complex operational problems, such as ad attribution, require sophisticated state management capabilities, which the transformWithState operator provides.</SPAN></P> <H3 class="c12"><SPAN class="c23">The Attribution Challenge: Correlating Disparate Real-Time Events</SPAN></H3> <P class="c15"><SPAN class="c5 c4">The Ad Attribution use case involves correlating two distinct, high-volume event streams: Ad Requests and Ad Impressions. These events arrive asynchronously, often out of order, and need to be stitched together across different time windows. Think of the "Request" event as an intention to show an advertisement at a given spot in a video or movie, and the "Impression" or ("ack") as the event that signifies that the ad is being shown to the viewer. Why is the "ack" needed? Primarily for these reasons:</SPAN></P> <OL class="c41 lst-kix_otsixhi2yo8s-0 start" start="1"> <LI class="c15 c45 li-bullet-0"><SPAN class="c5 c4">The ack payload may include additional intelligence (e.g., did the user click on or interact with the ad?)</SPAN></LI> <LI class="c15 c45 li-bullet-0"><SPAN class="c5 c4">Ad sales are often conditioned on an ad being shown to a specified number of viewers. Once that commitment has been met, the media provider has more ad slots to sell!</SPAN></LI> </OL> <P class="c15"><SPAN class="c4">For simplicity's sake, we are focusing on these two event types in the sample code. In the real world, there would be additional payload types to help with downstream attribution, such as associating a purchase with an ad impression. For our example, we focus on measuring impressions, as they are a key driver for managing ad inventory. Regardless, the technological challenge is maintaining the state of an active ad request, matching subsequent impressions and callbacks to that request based on complex logic (e.g., matching a request to an impression within 5 minutes, and then the impression to a click within 15 minutes), and ensuring accurate state expiration.</SPAN></P> <P class="c15"><SPAN class="c5 c4">Standard Structured Streaming operators, such as built-in stream-stream joins, can handle some correlation, but they often lack the flexibility required for arbitrary, time-driven state cleanup and complex object modeling inherent in sophisticated event correlation (CEC).</SPAN></P> <H3 class="c12"><SPAN class="c23">Why transformWithState is Mandatory for Operational Correlation</SPAN></H3> <P class="c32"><SPAN class="c5 c4">The transformWithState operator was introduced to enhance Structured Streaming with flexible state management and event-driven programming. It enables developers to implement arbitrary logic that goes beyond the capabilities of built-in operators. For operational workloads like Ad Attribution, several features of transformWithState are mandatory:</SPAN></P> <OL class="c41 lst-kix_l780j5j7apw3-0 start" start="1"> <LI class="c32 c43 li-bullet-0"><SPAN class="c28 c4">Object-Oriented State Management:</SPAN><SPAN class="c4">&nbsp;The operator allows the custom StatefulProcessor implementation to define and store complex, custom state objects (such as tracking the full history of a Request-Impression match object) rather than relying solely on simple key-value mappings.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">Composite Data Types:</SPAN><SPAN class="c4">&nbsp;Complex correlations often require composite keys (e.g., combining a User Session ID with a Campaign Identifier) to ensure that all related events are processed by the same state instance. transformWithState handles these composite types efficiently.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">Timer-Driven Logic:</SPAN><SPAN class="c4">&nbsp;This is the most crucial feature for attribution. Timers allow the system to register a future action, such as cleaning up state exactly 15 minutes after an impression fires, even if no further event arrives. This timer-driven logic ensures timely state garbage collection, which is essential for maintaining memory bounds and guaranteeing accurate attribution boundary enforcement in a high-volume pipeline.</SPAN></LI> </OL> <P class="c14"><SPAN class="c5 c4">The application logic and the execution performance (RocksDB health) are thus intrinsically linked. If state is not aggressively expired, the total state size grows, increasing I/O demands during checkpointing and taxing the bounded memory manager, thereby compromising the RTM latency goals.</SPAN></P> <H3 class="c12"><SPAN class="c23">StatefulProcessor Implementation Deep Dive</SPAN></H3> <P class="c15"><SPAN class="c4">The ad attribution architecture dictates a conceptual structure for the custom StatefulProcessor implementation. The input stream must first be grouped by a composite key (e.g., UserID and CampaignID) to guarantee co-location of related events at a single state instance. In the adtechdata_stream_processor_scala notebook, you will find this code that first groups the DataFrame by the transaction_id (think of this as an identifier for a viewing session) and then applies the transformWithState operator (LogRecordProcessor, a class that is defined in the same notebook):</SPAN></P> <LI-CODE lang="java">// Group by transaction_id and apply stateful processing // using LogRecordProcessor val transactionDF = sourceStream .as[InputRow] .groupByKey(_.transaction_id) .transformWithState( new LogRecordProcessor(timeout_duration = stateTimeoutInmin), TimeMode.ProcessingTime, OutputMode.Append ) .toDF()</LI-CODE> <P class="c15"><SPAN class="c4">The state management involves defining a complex object that tracks the progress of the attribution attempt. LogRecordProcessor is an instance of StatefulProcessor, which is required for transformWithState. When we look at the signature for LogRecordProcessor, a number of schemas are involved:</SPAN></P> <LI-CODE lang="java">class LogRecordProcessor(timeout_duration: Long) extends StatefulProcessor[String, InputRow, LogRecord] {</LI-CODE> <OL class="c41 lst-kix_6w69tmh79ob-0 start" start="1"> <LI class="c8 li-bullet-0"><SPAN class="c5 c21">String: The grouping key is a String. This will be the viewer's session identifier (in the source DataFrame, transaction_id)</SPAN></LI> <LI class="c8 li-bullet-0"><SPAN class="c5 c21">InputRow: This case class (also defined in the notebook) represents the schema of arriving data.</SPAN></LI> <LI class="c8 li-bullet-0"><SPAN class="c5 c21">LogRecord: The schema for outgoing records (emitted from the streaming query).</SPAN></LI> </OL> <H4 class="c49"><SPAN class="c56 c58">Event Handling Logic (Conceptual Flow)</SPAN></H4> <OL class="c41 lst-kix_7k2u59dzgcpg-0 start" start="1"> <LI class="c32 c43 li-bullet-0"><SPAN class="c28 c4">Ad Request Arrival:</SPAN><SPAN class="c5 c4">&nbsp;If no state exists for the key, the system initializes the state object, recording the request data and registering an Expiration Timer corresponding to the overall attribution window. In the sample code in the notebook, a timer is registered as rows for a grouping key arrive in the handleInputRows method. This helps keep the state data bounded over time.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">Ad Impression ("ack") Arrival:</SPAN><SPAN class="c5 c4">&nbsp;The state is updated, the impression is correlated with the tracked request, and the system may update or clean up state or emit output records.</SPAN></LI> </OL> <P class="c14"><SPAN class="c5 c4">The combination of RTM’s concurrent, low-latency execution and transformWithState’s precise, custom state control allows Spark to execute these complex, real-time correlation patterns accurately.</SPAN></P> <P class="c14"><SPAN class="c4">The stateful logic is encapsulated in the handleInputRows method of the LogRecordProcessor class, which extends the StatefulProcessor class, making it suitable for use with the grouped DataFrame's transformWithState method. Look in the adtechdata_stream_processor_scala notebook for this method:</SPAN></P> <LI-CODE lang="java"> override def handleInputRows( key: String, inputRows: Iterator[InputRow], timerValues: TimerValues ): Iterator[LogRecord] = { ....</LI-CODE> <P class="c27"><SPAN class="c5 c21">It is inside this method where complex, stateful logic resides. Two state variables (defined in the init method) are used:</SPAN></P> <OL class="c41 lst-kix_wpy1z8qtgfpe-0 start" start="1"> <LI class="c8 li-bullet-0"><SPAN class="c5 c21">_ackLogRecordListState: "ack" records are tracked in a list. It is possible for the "request" message to arrive late. By keeping ack records in a "list state" structure, once a matching "request" message arrives, the matching elements can finally be emitted. And, remember, if the "request" never arrives, the timer that was set for the grouping key will ensure that the system does not grow state data in an unbounded manner.</SPAN></LI> <LI class="c8 li-bullet-0"><SPAN class="c5 c21">_requestLogRecordState: this "value state" state variable tracks the "request" message and uses it when the "ack" messages arrive to finally emit the matched records.</SPAN></LI> </OL> <H2 class="c42"><SPAN class="c39 c53 c21">High-Performance State Management Internals</SPAN></H2> <P class="c15"><SPAN class="c5 c4">RTM's performance is dependent on the efficiency and stability of the underlying state store, particularly when running sophisticated stateful operators like transformWithState.</SPAN></P> <H3 class="c12"><SPAN class="c23">RocksDB Integration and Bounded Memory</SPAN></H3> <P class="c15"><SPAN class="c5 c4">The stateful Structured Streaming queries on Databricks rely upon the high-performance, embedded key-value store, RocksDB. RocksDB is optimized for flash storage and provides the I/O capacity required for high-volume, low-latency processing.</SPAN></P> <P class="c15"><SPAN class="c5 c4">A key challenge in operational streaming is preventing unbounded memory growth. RTM pipelines handle immense state pressure, and if not managed correctly, state stores can consume all worker memory, leading to instability. To prevent this, RocksDB functionality is leveraged to enforce bounded memory usage.</SPAN></P> <H2 id="h.ent35rkylt42" class="c42"><SPAN class="c39 c53 c21">Running the Demo: Comparing Microbatch to RTM</SPAN></H2> <P class="c15"><SPAN class="c4">The<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c28 c4">run_adtech_rtm_demo</SPAN><SPAN class="c5 c4">&nbsp;notebook contains everything you need to run the demo, which compares a streaming ad attribution stateful streaming query with RTM mode against the traditional micro-batch mode. However, you will need to change some configurations to use Kafka (or another supported streaming source/sink). The code includes detailed comments, and we'll cover some highlights here.</SPAN></P> <H4 id="h.r06u343v4x0g" class="c49"><SPAN class="c56 c58">Test Harness: AdTechStreamProcessor</SPAN></H4> <LI-CODE lang="java">import com.demo.data.{AdTechDataHelper} import com.demo.data.{AdTechStreamProcessor} // Get the active Spark session implicit val spark = org.apache.spark.sql.SparkSession.active // Initialize helper and processor classes for adtech data val dataHelper = new AdTechDataHelper(kafka_options, payload, schema) val streamProcessor = new AdTechStreamProcessor(kafka_options, schema)</LI-CODE> <P class="c27"><SPAN class="c5 c21">AdTechDataHelper is defined in the adtechdata_producer_scala notebook and manages the generation of fake data for the stream.</SPAN></P> <P class="c27"><SPAN class="c5 c21">AdTechStreamProcessor is defined in the adtechdata_stream_processor_scala notebook. There you will find the class used for the stateful transformWithState operator (LogRecordProcessor), along with the class (AdTechStreamProcessor) that helps with the benchmarking by establishing the streaming queries. What is striking is that, from a code perspective, using RTM is very simple. AdTechStreamProcessor's processStream method includes this code snippet:</SPAN></P> <LI-CODE lang="java"> // Write the processed records to the output Kafka topic val query = outputDf.select($"transaction_id".cast("string").alias("key"), to_json($"value").alias("value")) .writeStream .format("kafka") .options(kafkaConfig) .option("topic", outputTopicName) .option("checkpointLocation", checkpointLocation) .trigger( if (mode == "trigger") Trigger.ProcessingTime("0 minutes") else Trigger.RealTime ) .outputMode("update") .start()</LI-CODE> <P class="c32"><SPAN class="c5 c21">The ".trigger" is the only code-related place where a developer needs to reason about using RTM or micro-batch mode. For the benchmark, the processStream method starts the queries and adds some timestamps used for measuring latency between the inbound and outbound Kafka streams.</SPAN></P> <P class="c32"><SPAN class="c5 c21">To recap, to use the demo code:</SPAN></P> <OL class="c41 lst-kix_w8vg5i7ha9y-0 start" start="1"> <LI class="c32 c45 li-bullet-0"><SPAN class="c5 c21">Start with the README notebook.</SPAN></LI> <LI class="c32 c45 li-bullet-0"><SPAN class="c5 c21">Adjust Kafka connection details in the util/common notebook.</SPAN></LI> <LI class="c32 c45 li-bullet-0"><SPAN class="c5 c21">Use the run_adtech_rtm_demo notebook to run the benchmark code.</SPAN></LI> <LI class="c32 c45 li-bullet-0"><SPAN class="c5 c21">Explore the other notebooks in the util folder to explore the code responsible for generating the fake data stream (adtechdata_producer_scala) and for providing the stateful streaming query(adtechdata_stream_processor_scala).</SPAN></LI> </OL> <H3 id="h.ukmvw63llx6" class="c54"><SPAN class="c23">Benchmark Results</SPAN></H3> <P class="c32"><SPAN class="c5 c21">The results of the benchmark will display in the run_adtech_rtm_demo notebook, like this:</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="craig_lukasik_0-1769790863056.png" style="width: 863px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/23489i9696FE158CB1739F/image-dimensions/863x427?v=v2" width="863" height="427" role="button" title="craig_lukasik_0-1769790863056.png" alt="craig_lukasik_0-1769790863056.png" /></span></P> <P class="c32"><SPAN>The benchmark results, shown above, are from a run using DBR 18.0</SPAN><SPAN class="c5 c21">&nbsp;with three worker nodes (Standard_F8):</SPAN></P> <LI-CODE lang="markup">{ "cluster_name": "RTM_testing (compute optimized)", "spark_version": "18.0.x-scala2.13", "spark_conf": { "spark.databricks.streaming.realTimeMode.enabled": "true" }, "azure_attributes": { "availability": "ON_DEMAND_AZURE" }, "node_type_id": "Standard_F8", "driver_node_type_id": "Standard_E8_v3", "num_workers": 3, }</LI-CODE> <H2 class="c42"><SPAN class="c39 c53 c21">Conclusions</SPAN></H2> <P class="c15"><SPAN class="c5 c4">Apache Spark’s real-time mode represents a significant architectural evolution, moving Structured Streaming from a platform primarily focused on analytical throughput to one capable of supporting low-latency operational decisioning. This capability is achieved through the fundamental shift to concurrent stage execution, facilitated by the ConcurrentStageDAGScheduler and the specialized MultiShuffleManager for in-memory data transfer.</SPAN></P> <P class="c32"><SPAN class="c5 c4">For technical audiences and developers building these mission-critical systems, successful deployments with stable, predictable latency should consider the following:</SPAN></P> <OL class="c41 lst-kix_fa17jzgqh6n-0 start" start="1"> <LI class="c32 c43 li-bullet-0"><SPAN class="c28 c4">Configuration:</SPAN><SPAN class="c4">&nbsp;Activating RTM requires a simple cluster-level configuration.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">Capacity Planning:</SPAN><SPAN class="c5 c4">&nbsp;The concurrency constraint—that total task slots must equal or exceed total tasks—prohibits autoscaling and demands proactive capacity reservation sized for peak parallelism. This requires continuous monitoring of input volumes to prevent violation of the sizing prerequisite.</SPAN></LI> <LI class="c30 li-bullet-0"><SPAN class="c28 c4">State Management Optimization:</SPAN><SPAN class="c5 c4">&nbsp;Implementing complex correlation logic, such as Ad Attribution, relies on transformWithState for custom timer-driven logic and timers. This application-level logic must be paired with infrastructure optimization, particularly the mandatory activation of RocksDB changelog checkpointing, to stabilize the p99 latency and prevent periodic I/O spikes during checkpoint events.</SPAN></LI> </OL> <P class="c40"><SPAN class="c4">By enabling RTM and adopting<SPAN>&nbsp;</SPAN></SPAN><SPAN class="c4">the recommended best practices, RTM enables organizations to consolidate their data platforms, eliminating the need for fragmented, specialized streaming engines, while consistently delivering low-latency performance for the most demanding operational workloads.</SPAN></P> Wed, 11 Feb 2026 00:36:42 GMT craig_lukasik 2026-02-11T00:36:42Z https://community.databricks.com/t5/technical-blog/apache-spark-s-real-time-mode-use-case-focus-ad-attribution/ba-p/146141 <P><SPAN class="appsElementsGenerativeaiAstAnimated" data-ast-node-id="1">Ready to go under the hood of Apache Spark™ Real-Time Mode? Dive deep into the architecture, mechanics, and cluster sizing requirements for ultra-low-latency operational workloads. See how<SPAN>&nbsp;</SPAN></SPAN><CODE class="appsElementsGenerativeaiAstInlineCode">transformWithState</CODE><SPAN class="appsElementsGenerativeaiAstAnimated" data-ast-node-id="3"><SPAN>&nbsp;</SPAN>powers complex, stateful ad attribution with sub-second performance. Read the deep dive—code and benchmarks included!</SPAN></P> Wed, 11 Feb 2026 00:36:42 GMT https://community.databricks.com/t5/technical-blog/apache-spark-s-real-time-mode-use-case-focus-ad-attribution/ba-p/146141 craig_lukasik 2026-02-11T00:36:42Z