Real-Time Mode Technical Deep Dive: How We Built Sub-300 Millisecond Streaming Into Apache Spark™

Key Takeaways

Good afternoon. So, uh, Jerry and I are from Data Bricks. Uh, thank you for coming. Um, today we're going to introduce a new feature in Spark to allow you to implement real time applications. Um, real time is a very popular word. Um I searched data plus AI submit and see how many sessions have real time in this title and uh you can guess there are 25 sessions have real time in this titles. So real time is a really popular word. Um so today we're going to talk about the spark new features. Um this talk is going to be a tech technical talk. We're dive into the technology how this one can be implemented. Um tomorrow there will be another session and Jerry and Kartik will give the session uh tomorrow at 3 o'clock. Um the session is called delivering uh subsecond latency for operation workload uh on data bras. Uh this is actually is not counted as a part of 25 um talks. It doesn't have real time name. Um you can search Jerry Pong for his name and you can find a session tomorrow. Um so today we're gonna talk about this this new feature. Um oh no which direction to go. Okay. So so first we're going to give some introduction to uh Spark streaming and then we talk about why we need to build this new mode and how we build it. Um and in the end we'll give some demo to prove that this is real thing we have implemented. Um so uh so first of all spark have streaming capability. Some people might not know or know. Um this is the definition in this in this uh official page saying structure streaming is a scalable and fault tolerant stream processing engine built on the spark SQL engine. So remember scalable and fall tolerance. Um is also mentioned you have several ofistic subsystem. It will process it incrementally. It means as soon as some data comes we will process it um continuously. It never ends. We keep going keep going new data coming we process it and we'll update data to continue to arrive. Uh we don't wait for some some the end or something. As soon as we have data, we will output it. Right? Just remember fault tolerance scalable incrementally continuously updating data in to arrive. Right? There's one word is not mentioning it at all. Not mentioning at all. This word is latency. Right? It didn't mention actually uh one how long it takes from input to output is it is um not mentioned. Um so let's look at um the latency looks like for for the current system. So some some brief introduction how this implemented. It is implement in the so-called microbatch architecture. In this way data is cuted into batches. Actually every batch is processes like a batch query. Um for each batch in the planning phase it figure out what data it need to process and in the execution phase it read that part of the data and then process it and output it. So what does what does this um latency looks like or say freshness um because of this architecture for every batch the data process is actually arrived mostly in the previous batch right because that's how when planning phase we can see the batch so what's the latency means the latency is at least a batch durations right um because it's process the previous batch data the worst case is arrived in the beginning of the previous batch and furthermore because most of the time the output is saturally generated in the end of the batch. So we sometimes see the batch duration to be maybe close to two batches. Why this is the end of the generate the data is outputed by the end of batch. Um the the one of the reasons is that um the batches are are are executed in stages and these stages are run in sequential order. Only one stage finishes we start another one we start another one. So data is only generating the last stage and there's also scheduling overhead um credation overhead. So that these all of these are counted as part of the each of the record. So um how to how to make this fast? One way is like we just just speed up everything scheduling stages. So every every back duration is like 100 milliseconds or the duration could be less than 200 seconds. Um the way we did is we created a new way called realtime mode. We rethink um we don't do this microbatch way anymore. Um this is feature is right now in uh public public preview. you can try out a lot of queries can be supported. Um you can talk to your account team and see how to try it out. Um we're also going to open source it. Uh we got proven by the uh spark PMC and uh in the following months we will open source it. You will use you can you will be able to use it in open source. Um so um so so in in this new mode we're targeting the ultra low latency or say freshness in another terminology. In microbatch mode, we we usually targeting to a second level. Um and the real time mode, we are we're targeting a subsecond level to be 100 millisecond or even lower. Um so these applications will be um uh will be useful for some use cases such as like like internet of things and like fraud fraud detection or some of the the online applications. Um again you can come to tomorrow's session Jar and Cararthy will give some introductions and how this these things can be solved real world problems. Um so basic idea about this one is rather than try to shrink this batch small smaller we make the batch longer so that the scheduleuler overhead is is is arotized but on the other hand the data is streaming out inside of batching without batching at batching on end. So this some of the basic architecture. So uh remember the previous graph we process the uh the the the current batch process data from the arrived from previous batch in this cases a batch running very long and then this is this most recent data is is process uh uses comes and then it's it's flow all the way to the process process process engine and then get to the output. Um so we run the batches in fixed long durations usually minutes and by default five minutes and the data can flow from the source to the end without any buffering in the middle. So there are there are there there are some major changes we make to this architecture. Um first one first one is we need to we need to turn this stages from sequentially scheduled they are they are concurrently scheduled at the same time. Um we modify the source so that instead of read as fixed in fix the ranges we would read the freshest data as it comes and then we modify the the shuffling so that data is shuffled um as as it comes and uh we also uh schedule the st gauges in in concurrently. Um in order to achieve that um we need to change the way we do um checkpointing. Um in microbatch mode the checkpointing is done as in the planning phase of a batch. Uh we determine the the data we would like to process by looking at start and end offset and they are written to the checkpoint and and then we do the execution and then in the end up in the commit phase we write to a marker saying this is committed. Um if the query is restarted before it's committed um we will replay this batch in the same way um by looking at the logs and then if the query restarted after commit and this is considered done and would not be replay um but in a in a new mode uh when when we started batch we don't know we we don't determine the end of the batch we just keep processing all of this freshest data only in the end of processing we look at what offset we have process two and then at that time we know this batch is started and and offset and at that time we write to the checkpoint location and together with the commit log. So at at in in in this approach the only in the end we know the start and uh end of offset but this approach creates a problem because um in the previous previously a batch is a dumpotent no matter how many times you restart it it will it will um process the same amount of data. So that stay store checkpoint is a dumpotent. Um right now uh in in the new approach um in the real time approach uh stay sore checkpoint is not a dump anymore. Um so in order to resolve the problem we create a new way to do st checkpointing. Every checkpoint will be attached to a globally unique yoyo ID and then that ID will be committed um will be written in the commit log so that we know that which one is actually committed so there's no confusion. So putting thing together we made some major changes to um checkpoint mechanism to make it possible. Um we we changed this offset writing from the beginning of the batch to the end of the batch and then we also change the state source checkpointing to have a globally unique ID um so that it could be um attached to the actual commit. Um but by by the way this one we we we call checkpoint one v2 is also be a good thing for microbatch mode too and in the future we will we will also make it default to the microbatch um uh only make this shuffling and a source rating um uh in in streaming order sometimes is is is not enough. Sometimes operators also need to buffering. If we don't change that the the we the operators will only out output in the end of the batch and will be slow. So we need to modify some of the operator so that data are streaming. Um so the the one we modified this uh so far is aggregation and some of the UDF um operator and also there's there's transformer state um operator. This is new. I think there's a talk today. I hope some of them join that one. Um that feature by the way match this road time mode very well and then we also implement um the uh real time mode support for this this feature. So also there's a problem for um state cleanup um you know state store is the storage to to keep track of data across all of this processing even if the the the data is the query is restarted um in the past. Um but the story you cannot keep data forever. you have to clean it up after after a while if not useful. Um in the previously we clean up by in the end of every batch we scan the host store and then see whatever it is need to be clean up or just delete it. Um this is of voice will introduce long latency in real time mode. Um in real time mode we kind of doing this incrementally. We spreading this um cleaning up overhead to the whole five minutes. Um after processing every row, we we scan stay sort for several rows and see whether there's anything to delete and if there anything to delete, we delete it and then we process another row and then we we do another we scan stay for another several rows. By doing that one we can spread out this stay sort clean up job across the five minutes so that it doesn't impact individual rows latency. Yeah. So that's the summary what we did um we did this as a long timebased execution. Uh we scheduled the stages concurrently. We implemented uh source so that it it read the most um uh fresh data and then there's a streaming shuffle. That's major change to the to the shuffling so that it can stream in the data in shuffling and then we change the way we do checkpointing by changing the offset from the beginning to the end and then we change the way we do state store checkpointing to have unique ID and we also change the operators to reduce buffering and do the incremental um state cleanups. So this is the this is one again this is a available in public preview talk to account team to um um to try it out and uh it will be open source soon and I think the Jerry is going to give a preview. Hi everyone. I am Gary. Is everyone still with me or everyone still? Okay. So now let's talk about kind of the API that um allows you to use real time mode. It's super straightforward, super simple. Currently when you run a query structure streaming query, it looks like that, right? You have your read stream. You have maybe some transformations. You have your right stream which is you're writing it out. In this case, you're reading from Kafka, writing to Kafka. And then maybe you specify the kind of trigger type here, processing time trigger, which is the default one as well if you don't specify anything. So how do you enable real-time mode? That's it. You change the trigger, you get lower latency. And I will show that in the demo as well. And that's really all it is in real time mode for the user is basically switching the trigger. And that's one of the, you know, things that we intentionally built into real-time mode to encourage users to use it is basically the seamless user experience that you're able to switch between, you know, processing time trigger, trigger available now, and real-time trigger. We understand that users, you know, like yourself like have different needs often, right? Sometimes you want to have cost optimization. You want to have and prioritize throughput. But you know maybe something changes in the future. Right now you want to more focus on latency. You can change basically your trigger from say trigger available now which is more like a periodic incremental processing. You run it every so often but it's very cost effective. But you want to run that now with low latency. Simply change the trigger. So let me go over kind of the uh processing semantics. That's one of the you know interesting topics as well of real-time mode. Pretty straightforward as well. We guarantee the same processing guarantees as the existing kind of execution modes like you know the microbatch mode right same exactly once processing guarantees is maintained. You know we can get into a whole talk about what exactly is exactly once but you know here is a you know best attempt at summarizing that is each input record affects the final output exactly once. Right? So that is maintained. Let me talk about end toend guarantees. So for end to-end guarantees or exactly once end to end, exactly once delivery, however you know you want to kind of uh describe this, it's um it's really sync dependent as well, right? You have to kind of have support in your external sync to basically get that exactly once. often is you know we have some transactional support or you're able to do component rights for real-time mode we currently support the sync that we support is Kafka and for each syncs which for each sync is allows you to it's an interface an operator that allows you to write essentially wherever you want to write to you're given a row go write it somewhere wherever you want to do you're allowed to basically create custom syncs that way of course you're also able to use the you data source v2 API if you want to have the full breadth and depth of APIs to write a very you know complex sync you can also do that as well um end to end exactly once if you're interested it can be guaranteed right but you obviously have to um be able to write to that sync in an item way as well so it can be done uh but there's some extra work that might need to be there but for Kafka I mean the existing existing execution modes supports exactly once I at least once uh as well. So we're not really changing the guarantees there. So let's talk about performance, right? We're real we're talking about real time mode, right? We talk about it delivers ultra low latency. How do we actually compare with you know kind of the existing execution modes there, right? So here is a graph we've ran uh se several kind of benchmarks of common kind of patterns or queries that uh we have seen. To be honest, these are actually old numbers. The newer numbers are actually better. If you go to the keynote on Thursday, Michael's keynote, he will actually share some newer numbers hot off the press that are actually even more impressive. But the general idea is we see two orders of magnitude improvement in latency with real-time mode. All right, let's um go over the demo, which is honestly like the most exciting part for me. So, you know, if you were sleeping through the whole the the talk before or you're trying to push a commit, like this is the this is the part to really, you know, listen to because like it really demonstrates kind of the, you know, power of real time mode and it really allows you to visualize exactly what we're talking about here. So, let me first go over kind of the scenario of the demo. uh it is uh the demo is a continuous environ environmental monitoring scenario. So imagine a city where there sensors I mean it's not very hard to imagine I would guess nowadays that continuously monitors environmental conditions across different locations and zones. Each sensor collects measurements such as say temperature, humidity, CO2 levels, you know, part particular matter levels that are, you know, 2.5 micrometers or, you know, high like uh lower. And what the city uh administrators want to do is one say multiparameter threshold monitoring you know real-time detections of basically uh of of detection of violations of the of these uh sensor readings such as temperature, humidity, CO2 levels and so on and so forth you know around the city right another thing could be real time you know alerting right immediate notification when there are changes to an environment so city administrators you know can react to those situations accordingly in a timely fashion, right? Another thing could be monitor real-time trends, right? In analytics, you know, what are the trends are temp are these measurements rising? You know, uh decreasing, you know, what you know, what is going on with basically these sensors, right? So, let me also go over kind of the pipeline I have crafted. We have crafted um for basically a demo. Um we're basically using Kafka to Kafka. The source topic is Kafka, the sync topic is Kafka. Imagine measurements are being basically gathered from all sensors and published to a Kafka topic, right? They're read by an Apache Spark query. Um, it's a two-stage query. The first stage, you can think about it as a map stage, parses the data, right? And then we do a group by location because we want to do some analytics on a per location or per city basis. And we use the operator here transform with state as seeing mentioned there was already a talk today about it. It's our new basically custom state API that basically allows you to do powerful you know stateful transformations uh with with the data and then with that we can basically send alerts trends and aggregations uh to basically your downstream Kafka topic that then you can you know use to send out alerts power dashboards so on and so forth. So let's um you know get to the demo. I guess we can play it. The demo to showcase real-time mode and structure streaming. This whole demo is implemented on a data bricks notebook. Let me first walk you through the setup. I have already configured a Kafka cluster to be deployed that is reachable by this notebook. I have also started a data generator to generate data uh to be sent to the source topic of this Kafka cluster. we are using. First, let's inspect the data. What is the shape of the input data that we are going to process in this demo? Maybe I should have come up with a joke. It takes a little bit to, you know, it's real because, you know, it's actually trying to load it here. You should add it out. Uhhuh. or or speed speed it up, right? Yeah. Yeah. We're good on time. So, you know, we can This is not real time though. Is being continuously generated into the source topic. As you can see, it is these are the columns. There is a key and there is a value and then there's some metadata information such as such as topic, partition, offset, time stamp and so on and so forth. But let's take a look at the uh value here. This is where actually the sensor data that I mentioned before is shaped. There is a sensor ID. There is a location. There are temperature readings. There is humidity readings. There are CO2 readings and also particular matter levels as well. And each one of these readings has a key. And we're basically continuously sending this data into the source topic to be processed by our query shortly. So let me stop the query that reads from the source topic and go to the query we have here to actually process the data and send out meaningful alerts and analytics. So I've defined some data structures that my query is going to use. Here's a case class that describes the input the shape the data shape of the input sensor ID location city and so on and so forth and the different you know measurements that the sensors will uh provide. I have another case class here that is state. This is basically data that we are going to persist into the state store of transform of state to keep track of a historical list of data that we can use for analytics such as trends and averages and other sort of aggregations. Here is the output and this is what we're going to output from the query. It contains the sensor data, the input basically the row input of the query and also kind of the um the the analytics, some of the aggregations such as daily average, temperature trend, high temperature count and any sort of alerts that have that that are produced in which any of these kind of readings such as temperature, humidity cross a certain threshold and we want to alert on those high readings from the sensors. And here is the definition of the stateful processor. Really the meat of the query here. We're using transform of state. And really the query like I alluded before is simply storing historical um storing rows of data to calculate basically trends and aggregations. and also based off of these thresholds to send out alerts if c certain readings are above a certain threshold. So we can see the code here you know we're storing basically uh historical data. We are also using that to analyze trends and subsequently we are basically checking the current temperature to see if there are any sort of alerts that we need to send out and we are outputting you know our data here. Let me actually run these and then oh let me run this again and now let's actually run the query here that I have defined that reads from our source topic and writes to our sync topic here that uses that reads that basically deserializes the JSON passes to the transform and state operator that we're using for our alerts and analytics and writing it to a Kafka topic here. First, let's use the existing processing time trigger that already exists uh today for structure streaming. And even if you don't set a trigger, this is basically the trigger that it will default to. So, let's wait for the query to run. Let's go over to another notebook that can basically read out the results. So I have a notebook here that's basically reading from the sync topic and it's basically going to output uh what is written in the sync topic as well as latency that um the record took that spark took to process the record. And how do we define latency here? We define latency as the time the record was first persisted into Kafka or the log append time of the of each input record and the source topic and also the log append time of each record when the structure streaming query writes to the sync topic. So we're really canu computing the time that has elapsed for this record when it was written to the source Kafka topic and when it was run to the sync Kafka topic. So let's run the query. As you can see we are getting the results. Now we have a s each row has a sensor ID location city time stamp the temperature readings that were part of the input but now we're also computing a daily average temperature which is basically using the data that's persisted in the state managed by the engine via transform of state. It also computes a temperature trend whether the current reading contributes to a failing, a falling, a stable trend or a rising trend. Here's the number of actually high temperature counts that we've computed as well. And some alerts such as for example this alert has a high CO2 alert. You know uh also uh a high particular matter alert so on and so forth. And the last column here is latency which I want to talk a little bit about more. As you can see the current latency for processing each one of these records you can see is about you know around 1 second. Right? This is the latency we're getting with the existing processing time trigger or the existing kind of execution mode that produces the lowest latency possible in the existing engine. Now let's simply switch over to using real time mode. Let's restart the query. Start it again. So now all we really need to do for starting the query in real-time mode is simply change the trigger and you can seamlessly switch between other processing modes in structure streaming to realtime mode. Okay, the query is running. Let's switch over to see what the input is. So now you can see the input and the existing triggered processing time trick uh processing time mode was around you know 1 second. we restarted. So there was some high latency but once we really transition to realtime mode we can see that the latencies are now in the doubledigit low double digit milliseconds. So compared to the existing processing time trigger we have basically two orders of magnitude uh difference here. The latency in real-time mode is two orders of magnitude smaller than the latency a that's able to be produced by processing time. And that's really the cool thing about real-time mode. You can seamlessly switch queries between different execution modes and immediately see the difference in latency. This is all I have for the demo. Thank you for listening in and hopefully this gave you a glimpse of the Okay. All right. So hopefully this gives a great demonstration of a real time mode. Um if you're interested like seeing mentioned one, please come talk to us about your use case. Number two, you can talk to your um basically account team of how you can get um onboarded to uh real time mode uh as well. And I guess we have time for Q&A.