Streaming joins are hard(estuary.dev) |
Streaming joins are hard(estuary.dev) |
We've been building such an engine at Feldera (https://www.feldera.com/), and it can compute joins, aggregates, window queries, and much more fully incrementally. All you have to do is write your queries in SQL, attach your data sources (stream or batch), and watch results get incrementally updated in real-time.
It is indeed inspired by timely/differential, but is not exactly comparable to it. One nice property of DBSP is that the theory is very modular and allows adding new incremental operators with strong correctness guarantees, kind of LEGO brick for incremental computation. For example we have a fully incremental implementation of rolling aggregates (https://www.feldera.com/blog/rolling-aggregates), which I don't think any other system can do today.
But i’d work in “just like Postgres”.
I've written my fair share of joins in SQL. They're indispensable.
But I've never come across a situation where I needed to join data from two streams in real time as they're both coming in. I'm not sure I even understand what that's supposed to mean conceptually.
It's easy enough to dump streams into a database and query the database but clearly this isn't about that.
So what's the use case for joins on raw stream data?
A lot of people just learning about streaming systems don't come in with useful intuitions about when they can and can't use that approach, or even that it's an option. We're hoping to build up to some documentation that can help new people learn what their options are, and when to use each one.
Kafka Streams, by default, uses either RocksDB or an in-memory system for the join buffer, which is fine but completely devours your RAM, and so I have been writing something more tuned for our work that actually uses Postgres as the state store.
It works, but optimizing JOINs is almost as much of an art as it is a science. Trying to optimize caches and predict stuff so you can minimize the cost of latency ends up being a lot of “guess and check” work, particularly if you want to keep memory usage reasonable.
Streaming joins are so hard, that they're an anti pattern. If you're using external storage to make it work, then your architecture has probably gone really wrong or you're using streams for something that you shouldn't.
I view batch tables as simply a given state of some set of streams at a point in time. Running the same query against "batch" tables at different points in time yields different results (assuming the table is churning over time).
I don't really see the difference between tables & streams. Data in tables changes over time too. You can model a stream as a table with any degree of fidelity you desire. In fact, I believe this could be considered a common approach for implementing streaming abstractions.
At best, it is a trick to be applied in very specific circumstances.
You want to end up with the results of:
``` select * from clicks left join impressions on (clicks.impression_id=impressions.id) ```
but you want to see incremental results - for instance, because you want to feed the joined rows into a streaming aggregator to keep counts as up to date as possible.
I was definitely under the impression that ad impressions and clicks would be written to databases immediately and queried from there.
I'm still having a hard time imagining in what case you'd need a "live" aggregating display that needed to join data from multiple streams, rather than just accumulating from individual streams, but I guess I can imagine that there are circumstances where that would be desired.
Thanks!
Let's say you run a large installation that has a variety of very important gauges and sensors. Due to the size and complexity of this installation, these gauges and sensors need to be fed back to a console somewhere so that an overseer role of sorts can get that big picture view to ensure the installation is functioning fully healthy.
For that scenario, if you look at your data in the sense of a typical RDBMS / Data Warehouse, you would probably want to save as much over the wire traffic as possible to ensure there's no delays in getting the sensor information fed into the system reliably on time. So you trim down things to just a station ID and some readings coming into your "fact" table (it could be more transactionally modeled but mostly it'll fit the same bill).
Basically the streaming is useful so that in near-realtime you can live scroll the recordset as data comes in. Your SQL query becomes more of an infinite Cursor.
Older ways of doing this did exist on SQL databases just fine; typically you'd have some kind of record marker, whether it was ROWID, DateTime, etc., and you'd just reissue an identical query to get the newer records. That introduces some overhead though, and the streaming approach kind of minimizes/eliminates that.
What I don't understand is streaming joins. None of your gauge values need to join to anything.
And if they did -- if something needed to join ID values to display names, presumably those would sit in a database, not a different stream?
If you have distributed data, the join will get calculated by SOME node in the network, and the data will have to be streamed in and joined by the central processor. Even with modern meganodes, for BigData marketing you have to handle arbitrarily sized datasets, and that means streaming data into the processing nodes working memory.
Of course there are ways to distribute join calculation (sometimes) as well, but you're still talking merging streams of data coming into processing nodes.
Now, if you have to handle AP/eventually consistent models, then it REALLY gets complicated, and ultimately your huge massive join (I'm assuming a join of tables of data, not just a denormalization join of a single row/primary key and child foreign keys) is a big eventually consistent approximation view, even without the issue of incoming updates/transactions mutating the underlying datasets as you stream and merge/filter them.
There's also cases where you just want real time results. For example, if you want to take action based on a joined result set, then in the rdbms world yoy might periodically run a query that joins the tables and see if you need to take action. But polling becomes increasingly inefficient at lower polling intervals. So it can work better to incrementally compute the join results, so you can take action immediately upon seeing something appear in the output. Think use cases like monitoring, fraud detection, etc.
In a very large number of cases, those streams of data are too large to query effectively (read: cheaply or with low enough latency to satisfy people interested in up-to-date results) at rest. With 100ks or millions of events/second, the "store then query" approach loses fidelity and affordability fast.
How important ads are is debatable, but if you're an ad company and this is what your customers want, it's an implementation that you might come up with because of the engineering practicality.
At a high level the push-instead-of-pull benefit here is "you don't have to query the ID values to get the display names every time" which will reduce your latency. (You can cache but then you might get into invalidation issues and start thinking "why not just send the updates directly to my cache instead")
There's also a less cacheable version where both sides are updating more frequently and you have logic like "if X=1 and Y=2 do Z."
For small enough batches streaming and micro-batching do often end up very similar.
"Note that the SQL ORDER BY directive can be modelled as a non-linear aggregate function that emits a list. However, such an implementation is not efficiently incrementalizable in DBSP. We leave the efficient handling of ORDER BY to future work."
My understanding is that Feldera does indeed support ORDER BY, which I imagine it does incrementally, thus my question.
The statement in the paper that ordering is not efficiently incrementalisable seems to makes sense to me. It is clear that even though Z-sets are not naively able to represent diffs of ordered relations (since Z-sets are unordered), ordering can be modelled as an aggregate that first builds up the first row, then the second row, and so on. Even as formulated this way however, I fail to see how the entire "incrementalised" computation would still be practically incremental, in the sense that the size of the output diff (Z-set) is small as long as the diff of the input is small.
For example, consider the query `select x from y order by x asc`, and the following values respectively occur in the stream of x: 5, 4, 3, 2, 1. Now, consider the incremental diff for the last value of 1. If presumably one models order by a list aggregation, then the Z-set for the entire computation seems to be
- [ 2, 3, 4, 5 ]
+ [ 1, 2, 3, 4, 5 ]
- 2 (index: 0)
- 3 (index: 1)
- 4 (index: 2)
- 5 (index: 3)
+ 1 (index: 0)
+ 2 (index: 1)
+ 3 (index: 2)
+ 4 (index: 3)
+ 5 (index: 4)
So, a better solution is still future work :)
The reason DBSP and Differential Dataflow work so well is because they are specialized to relational computations. Relational operators have nice properties that allow evaluating them incrementally. Incremental evaluation for a general purpose language like Rust is a much, much harder problem.
FWIW, DBSP is available as a Rust crate (https://crates.io/crates/dbsp), so you can use it as an embedded incremental compute engine inside your program.
For how elegant DBSP is I still found the paper a tough nut to crack, and it really is one of the more accessible theoretical contributions in the space, at least from this grubby programmer's perspective... I hope to devote some time to study and play around more, but in the meantime I'm rooting for you!
You may want to check out this tutorial for a hands-on introduction to DBSP: https://docs.rs/dbsp/0.28.0/dbsp/tutorial/index.html