How Photon Unlocks New Performance Gains for Streaming

Summary

Databricks' Photon vectorized C++ engine is being extended to stateful streaming operators that previously relied on the JVM execution model, closing a long-standing performance gap for streaming workloads.
Benchmarks demonstrate a 3x or greater speed-up on simple streaming ETL and a 2x speed-up for Photonized drop-duplicate operations, translating to lower latency, higher throughput, and reduced Total Cost of Ownership.
The engineering roadmap includes expanding Photon coverage to aggregations, stream-stream joins, and the Kafka source connector, with Photon available today on both serverless and dedicated clusters.

How Photon Unlocks New Performance Gains for Streaming

Databricks’ Photon engine has revolutionized Spark execution, delivering state-of-the-art query performance and cost efficiency. While many Structured Streaming workloads already benefit from this vectorized execution engine, a subset of streaming-specific operators has historically relied on the standard JVM execution model, leaving potential optimization on the table. Until now. In this video, we reveal our recent engineering initiatives to expand Photon coverage to these critical streaming operators. We will dive into the technical journey of “Photonizing” these components, detailing the complex engineering challenges we overcame to bring vectorized execution to stateful and streaming-specific logic. Attendees will see side-by-side performance benchmarks demonstrating how these enhancements translate into lower latency, higher throughput, and reduced Total Cost of Ownership (TCO) for streaming pipelines
Talk By: Sagar Mittal, Software Engineer, Databricks ; Siying Dong, Senior Staff Software Engineer, Databricks ;

Chapters

FAQs

What is Photon in Databricks?

Photon is Databricks' vectorized C++ execution engine that uses SIMD instructions to process data in column batches, delivering state-of-the-art query performance and cost efficiency. It was originally designed for batch workloads and is now being extended to cover streaming-specific operators in Spark Structured Streaming.

Which streaming operators are being Photonized?

Databricks engineers are expanding Photon coverage to stateful streaming operators that previously ran on the standard JVM model, starting with drop-duplicates. The roadmap includes aggregations, stream-stream joins, and the Kafka source connector.

What performance improvements does Photon bring to streaming workloads?

Benchmarks show a 3x or greater speed-up on simple streaming ETL workloads and a 2x speed-up for Photonized drop-duplicate operations. These gains translate to lower latency, higher throughput, and reduced Total Cost of Ownership for streaming pipelines.

How do I enable Photon for streaming on Databricks?

Photon can be enabled on both serverless and dedicated clusters in Databricks. Once enabled, streaming workloads automatically benefit from vectorized execution wherever Photon coverage is available.

Full transcript

[00:07] Hello everybody. Um I'm Siying from Databricks stream engine team. And this is my teammate Sagar. Um today we're going to talk about Photon and how it can be can improve
[00:20] performance of streaming. Um This is standard slides. We're going to talk about um type of streaming queries and what type will be um
[00:32] what what demand better cost efficiency and some brief introduction what Photon is and whether streaming can benefit from Photon and how we narrow the gap.
[00:49] Um so first we'll talk about what kind of streaming queries there are. Um the streaming capability in Spark is called Spark structured streaming. And in structured streaming we define
[01:02] streaming slightly different from some other systems. We define streaming queries as the source of the query is unbounded. It's keep coming, keep coming. And then we we incrementally
[01:15] process the data since the last time. Um Note that we don't talk about continuously streaming query. It don't have to run continuously and we
[01:27] don't talk about that the data are generated as soon as comes or streaming out. Um it's it's only about it source is unlimited and the processing is incremental.
[01:39] Um in fact many of the streaming queries uh Spark streaming queries are running hourly or even daily. Um you may wonder that if the job of streaming query is running daily, what's the difference between um the streaming
[01:52] query and a batch query? So um the the but uh um so these are the reasons people run streaming query hourly or daily cut because they provided two additional
[02:05] things compared to a batch query. Um it remembered offset. Um that means that after we recover from next run, it will pick up what's left from the last one uh the last run and
[02:18] then run out all the newer outstanding data. So, you don't have to deal with it daily the the partition day partition, hour partition, and then deal with
[02:30] um the late data, when to trigger the job, everything is automatic for you. And there's there is another thing a state store because sometimes you would like to uh to keep some state across this run.
[02:44] For example, if you have duplicate data, you would like to keep some of the the previous data in a state store so that after it's wake up again, you can do deduplication from the previous run.
[02:58] Um so, we we we talk about the the the the streaming jobs could be daily or hourly. So, streaming jobs so so that streaming job can be organized in a spectrum. In the very left side, uh there there are
[03:10] use cases demanding latency. Um those those are the cases that uh they would like the the query to to act to take action for each event immediately. The these most of
[03:23] them what do what do what do we call operational workloads? Um they these queries uh are latency sensitive and they usually cannot be very cheap because the
[03:37] hour has to be keeping running and then we cannot batch much and there's constant cost even if there's not much data coming. Um on the very right side, um these are the trigger job, daily or hourly jobs.
[03:52] These are mostly ETL jobs. People would like to move data to data warehouse to to data bricks and then do some transformations. They the cost is the main concern. They would like to do this
[04:04] operation in the most cost-efficient way. So, from the lower latency end, we have a feature called real-time mode.
[04:16] The um I believe tomorrow 3:00 p.m. there's a session about real-time mode on feature engineering. My teammates would give a talk on it.
[04:29] And on Thursday, there will be a three talks by three different users on how they take advantage of real-time mode and then to to solve their business problems. And this talk is is not about
[04:43] that. Um regarding to low latency workload, these are like some example of all this workload including some fraud detection, some security monitoring, and
[04:55] it could do some the real-time machine learning features uh generation and then some of internet of things. But this is not this talk is mainly about.
[05:07] Um this talk is mainly about the other end where the job might be triggered in higher interval, mostly scheduled ETL jobs. These could be some incremental ingestion data. You
[05:21] get data from Kafka, Kinesis, object store into the data lake. You do some simple transformation. You may do some data enrichment, joining some data, doing
[05:33] some aggregation, deduplication. Um these are the use cases that usually cost is very important for them. And then we have two side of them,
[05:45] optimized for performance, for latency, and optimized for cost. So, uh what DataBricks DataBricks provides to improve cost for this use cases,
[05:58] um there are two set of things we provide to improve cost. One is serverless. With serverless, um there's auto scaling um to just scale to the number of slots
[06:11] that are used as needed. So, that looks slot utilization can be high. There's less waste uh on the right side is what we're talking about today. It's the Photon. Or the goal of Photon is to use
[06:25] less hardware to serve the same query. Right. So, for people who are not familiar with Photon, it's a brief brief introduction about
[06:37] the what Photon is. Photon is native C++ engine. Um the some of the operators Spark operators is re-implemented in using C++.
[06:50] Um and the scheduler is also re-implemented. And it's heavily rely on columnar and vectorized processing, and then use SIMD to speed up.
[07:03] And it's supposed to be very transparent to code. The same code can be run in Photon and non-Photon mode. And you don't need to any changes as long as the operators are supported, it will
[07:15] automatically photonize and and realize this gain. And the question about whether streaming queries can benefit from Photon just as batch queries,
[07:27] um of course, yes. So, um by um the good thing about Spark structured streaming is it will reuse lots of operators in batch queries.
[07:39] Um the simple queries like reading and reading from parquet files or writing to parquet files, doing simple projection, filtering, and joining with a reference table. These are the
[07:52] the code is mostly shared from batch queries and then photonization it is is working for both. All right. Um, we actually do some measurement. We measure the four queries, um, which is
[08:04] pretty common. One is just simple pass through. We just just just mirroring the reading from parquet file and writing from parquet file. And another example would be we just reading from parquet file and then join with another, um,
[08:17] table and then and then and then writing out. And another case is we joining two tables and writing out. And then finally there's some demultiplexing table and then we getting from a table and then based on some content from each column
[08:30] and splitting to multiple tables. All right, these are all common ETL queries. The result shows that um, we see mostly three times speed up or higher for those queries.
[08:43] So, Photon does work for quite a lot of queries in, um, in streaming queries. So, then what's the gap? Um, there are some gaps, um,
[08:57] that we should be easily filled by, um, photonizing some operations. But there are some gaps what we thought would be very difficult. Um, that's why we got there to experiment it. Um,
[09:11] so, the stateful operator is what we thought would be very difficult difficult one to be, um, to be photonized. As as as what what we said, streaming queries is is uh provide
[09:23] something called state store to keep state across multiple batches. Um, so the simple thing is there's there's operator comes and this operator have some RocksDB state store, can read and write, and then this is or is
[09:36] persistent in in in the checkpointing so that is is it can be reloaded or we keep going in the next batch. Um but the state store actually is a very
[09:49] very complex system. Um so it's not just having state store running in local RocksDB. We have to do the checkpointing there. And the checkpointing by itself is very complex thing in order to to to do it
[10:03] very efficiently. And then we have to manage older version of checkpoint. We have to manage the thing about the loading from checkpointing and and then manage the thing.
[10:15] And and and and there's also important thing is that we have to make sure that affinity of those those tasks so that we we assign the task to the checkpoint to
[10:28] to the to the existing state store so that we don't have to reload from cloud. So the whole thing is a very complex system. It makes very hard to photonize
[10:40] because first of all, the whole thing is very complex. So we have to even if to photonize one single state for operator, we might need to rewrite everything in C++. And even if we rewrite everything in C++
[10:53] it create problem about we have two set of state store management systems and they have to co-exist inside of the same executor because some queries photonizable some not. Even the
[11:06] same query, some operator might be photonize some might not. Even the same query, you might switching from photonize non-photonize based on changes. So that's very complex and then most
[11:18] importantly and because they store is persistent thing, you have to make sure that it can be switchable between photon and non-photon version.
[11:30] Photon and non-photon version. So this is make things very hard, but we would like to explore if there way we can fortonize it the stateful operator. If we can do this, we feel like um most of the operators should be fortonizable in
[11:43] streaming queries. So, that we we picked the one of the stateful queries, drop duplicates to fortonize, and then and then we succeeded. Um for so so this is how drop duplicates
[11:57] does. When there's input rows coming from this operator, we look at a state store and see whether this there's existing um row in the state store for this input record. If there's already one, we just
[12:10] ignore this row. If it doesn't exist, we insert this row in the state store and output. So, in this way we we take the input and only output one per um per per
[12:22] row um and dedupe. So, this is the idea we decided to to do it to solve all the problem of complexity. We actually
[12:35] keep all those state store management in JVM and uh only do this this this execution part in C++. And then communicate these two parts through JNI.
[12:52] Um and uh and these two parts share the same RocksDB, and the JVM part manage it. Um and and the and the and the C++ part only only gather the um
[13:05] the the handle of the RocksDB and then do the actual the row processing and then the read and write from RocksDB and emit this change log bytes back to JVM. So, that you can see that um
[13:17] we would try to um do most of these um the evaluation code in in C++ and then keep all the management code in JVM. So, to uh um to this is how the the the
[13:32] workflow that it will be a shared RocksDB and the managed it by JVM for the whole life cycle Open a DB do the commit, checkpointing, and the maintenance, and the matrix is all done
[13:45] by JVM. And when there's new query coming, the code compilation, query planning is still done in JVM just as just as all other queries and then generate the vectorized operator if it's
[13:58] if it's qualifies and then it handle to the Photon part to to to to do the the execution. Uh in the beginning of the task, um it asked the JVM part for the RocksDB
[14:10] handle and the RocksDB handle um is passed to the JVM part so that they can do get and get and put. And uh for each column batch and Photon's pre-processing column batch
[14:22] batch by batch for for for for each column batch, it's is does some pre-processing of data um doing some this late event filtering and then and then do some deduplication
[14:35] within the um the batch and then it's it's directly called RocksDB in C++ function and then to get um to see whether the the the keys exist. For those key does not exist, it's it's
[14:49] construct what to write RocksDB and then just write to RocksDB. And uh and and then finally it's generated change log. It's kind of the delta log to do checkpointing and
[15:03] then the the generated bytes and then just pass it to JVM to actually write to the S3 file of the change log file. And then this batch finish processing and it's ready to to
[15:16] to do another batch and then it will do two to nine and again again until the end of task. In the end of the task, it will do the final call to the JVM side to
[15:29] um to commit the RocksDB and update its map the metrics. So, this is how the flow of um a deep deep dupe operation works with
[15:42] um Foton. So, why we think that's a good ones, right? There's There's most of the critical path in C++. Um, we do RocksDB read and write. All
[15:54] those read processing is in in uh C++ side. And then the state handling logic is just one. This is just only one state management inside of JVM, no matter how many queries are there, Fotonized and
[16:07] not Fotonized. There's just one management. There's no confusion there. And there's just only one owner of RocksDB, which is the JVM side. There's no two on own owner problem. So, we
[16:19] think the split is right one. And there's all There's only three JNI calls uh needed to to synchronize between JVM and non-JVM side. We need to
[16:31] get this RocksDB handle, and then we we pass the change log to to to do the change change log checkpoint, and then finally we do the checkpointing with the metrics.
[16:47] So, is it is important to make sure the data is correct. Um, this is especially uh important because we have persistent format on the for the state store, so it have to be correct. So, we do some more conservative
[16:59] key types. Uh, we only we only um open up this key key types if we validated we are it is working, and we will slowly open up some of the new ones
[17:12] if it's needed. And then we gated some of the options for Fotonization uh only open up if we fully test this supported. And we make sure that metrics are are exactly same based on our
[17:25] testing to make sure the behavior is the same. So, we fortunize drop duplicate case and then run the benchmark and then we show that this is 2x faster, which is very
[17:37] good results. The reason this is faster because there's there's several major reasons because most of quick lines are are
[17:49] already fortunized. And the the fact that state for our cannot be fortunized means we need to conversion from row to column and conversion back, which is expensive. It's also the execution itself is slightly faster. Some of them
[18:03] we rely on Cindy execution. And we can do because RocksDB is done it is in C++ so we are able to avoid JNI call and memory copy and object creation
[18:17] based on that. And also RocksDB's multi get is has more efficient interface with C++ with the raw pointers. So, these all contribute to the speed up
[18:30] reasons. Potentially we can unblock more saving if we do a more deeper optimizations. RocksDB is general purpose database that there's opportunity we we find we can
[18:43] speed it up if we have some customized implementation. And because we do C++ we should be able to to do some RocksDB compaction filter so that we can move the state eviction ones in out of the the critical path.
[18:57] And we can also do pre-store some record to make it friendly to to load the data into RocksDB. These we could be could be done if it's if if needed.
[19:11] Because we after we try out the first state operator, we know how to do this. So, we we are doing in more state operators including aggregations, stream stream join, and transfer state. So, these were
[19:25] all be done in the coming coming month or year. Also said, stateful operator is not only um, major operators is not photonized.
[19:38] There are still some pretty heavily used streaming operators that's not photonized. And there is Kafka source reading from Kafka we're still using the Java library. And then the the other
[19:50] readers parquet reader sometimes are not photonized. Um, and the deserialization libraries so sometimes heavily used are not photonized. So, we are working on photonized them all so that
[20:04] um, they will be able to um, be have this speed up um, similar to the one we have state operator or even higher.
[20:19] So, the final way how to how to enable Photon. Um, if you if you use serverless, both of serverless classic or serverless through Spark declarative pipeline, they're all
[20:33] always photonized. Um, you cannot even turn it off. Um, if you created your own cluster rather than using serverless, you when you create a cluster, you can go to the performance tab and then find the
[20:46] Photon acceleration check white check check box and then check that one. And then the cluster will be created with with with Photon and then all the queries will be photonized if possible. And there will be a public doc about for
[21:00] Photon you can go there and check. Okay, so this is what what I talked today. Um, Spark structure streaming have a range of
[21:13] queries starting from the latency end which you can use real-time mode and lower processing speed with a cost end where you can use higher trigger
[21:25] intervals and then Photon will help the the the the the latter ones most. And many streaming queries already benefit from queries from Photon. There's there
[21:37] are there are already see a good speed up if your query is like doing this this this triggering thing. The major gap I mean mindset technology gap is stateful
[21:49] operators and we Photon ice dropped your PKS as the first stateful operators and that's one
[22:02] and then we will we will do more stateful operator for the nation and then some other operators that commonly used in streaming queries. Yeah. That's that's all for today.
[22:14] >>

Learn more about the Databricks Data and AI platform.

The information provided herein is for general informational purposes only and may not reflect the most current product capabilities or configurations.