[HN Gopher] Streaming joins are hard
___________________________________________________________________
Streaming joins are hard
Author : danthelion
Score : 50 points
Date : 2024-10-22 14:35 UTC (3 days ago)
(HTM) web link (estuary.dev)
(TXT) w3m dump (estuary.dev)
| jdelman wrote:
| The ability to express joins in terms of SQL with Estuary is
| pretty cool. Flink can do a lot of what is described in this
| post, but you have to set up a lot of intermediate structures,
| write a lot of Java/Scala, and store your state as protos to
| support backwards compatibility. Abstracting all of that away
| would be a huge time saver, but I imagine not having fine grained
| control over the results and join methods could be frustrating.
| fiddlerwoaroof wrote:
| Flink does have a SQL join now that you can make work.
| Streaming joins remain a hard problem, though and, imo, SQL
| doesn't map nicely onto streaming systems.
| fifilura wrote:
| A couple of years ago Materialize had all the buzz, not sure what
| is the difference.
|
| https://materialize.com/
| neeleshs wrote:
| "Unlike batch tables, streams are infinite. You can't "just wait"
| for all the rows to arrive before performing a join."
|
| 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).
| lsuresh wrote:
| Your mental model is spot on and described quite well here:
| https://current.confluent.io/2024-sessions/streaming-queries...
| crazygringo wrote:
| Can someone explain what the use case is for streaming joins in
| the first place?
|
| 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?
| BeefWellington wrote:
| I'll use a contrived example here to explain what the value of
| streaming the data itself is.
|
| 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.
| hotstickyballs wrote:
| Should've just cached the output of group bys.
| crazygringo wrote:
| I definitely understand the value of streaming. Your gauges
| example is great.
|
| 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?
| closeparen wrote:
| Anything you can do with stateful streaming technology, you can
| do with a database and a message handler. It's just a question
| of programming model and scaling characteristics. You typically
| get an in-process embedded DB per shard, with an API that makes
| it seem closer to managing state in memory.
| tshaddox wrote:
| Isn't the use case just any time you want a client to
| essentially subscribe to an SQL query and receive message every
| time the result of that SQL query changes?
| GeneralMayhem wrote:
| Event correlations are a typical one. Think about ad tech: you
| want every click event to be hydrated with information about
| the impression or query that led to it. Both of those are high-
| volume log streams.
|
| 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.
| crazygringo wrote:
| That's helpful, thanks.
|
| 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!
| jrockway wrote:
| I think it can be challenging to get that much data to a
| single database. For example, you probably don't want to
| send every "someone moused over this ad" event in Japan to
| a datacenter in us-east-1. But if you do the aggregation
| and storage close to the user, you can emit summaries to
| that central server, backing some web page where you can
| see your "a 39-year-old white male moused over this ad"
| count go up in real time.
|
| 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.
| ryzhyk wrote:
| The computational complexity of running an analytical query on
| a database is, at best, O(N), where N is the size of the
| database. The computational complexity of evaluating queries
| incrementally over streaming data with a well-designed query
| engine is O(delta), where delta is the size of the *new* data.
| If your use case is well served by a database (i.e., can
| tolerate the latency), then you're certainly better off relying
| on the more mature technology. But if you need to do some
| heavy-weight queries and get fresh results in real-time, no DB
| I can think of can pull that off (including "real-time"
| databases).
| 10000truths wrote:
| Streams are _conceptually_ infinite, yes, but many streaming use
| cases are dealing with a finite amount of data that 's larger
| than memory but fits on disk. In those cases, you can typically
| get away with materializing your inputs to a temporary file in
| order to implement joins, sorts, percentile aggregations, etc.
| hamandcheese wrote:
| It seems intuitive to me that a correct streaming join is
| impossible without an infinite buffer and strong guarantees on
| how events are ordered. The number of real world systems offering
| both of those guarantees is zero. Anyone espousing streaming
| joins as a general solution should be avoided at all costs,
| particularly if they have a title that contains "architect" or
| "enterprise" (god forbid both in the same title).
|
| At best, it is a trick to be applied in very specific
| circumstances.
| ryzhyk wrote:
| A streaming join indeed requires an unbounded buffer in the
| most general case when inputs keep growing and any input record
| on one side of the join can match any record on the other side.
| However, it does not require inputs to be ordered. An
| incremental query engine such as Feldera or Materialize can
| handle out-of-order data and offer strong consistency
| guarantees (disclaimer: I am a developer of Feldera). In
| practice, unbounded buffers can often be avoided as well. This
| may require a specialized join such as as-of join
| (https://www.feldera.com/blog/asof-join) and some GC machinery.
| mattxxx wrote:
| JOINs are just hard _period_. When you 're operating at a large
| scale, you need to be thinking about exactly _how_ to partition +
| index your data for the types of queries that you want to write
| with JOINs.
|
| 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.
| tombert wrote:
| A large part of my job in the last few months has been in the
| form figuring out how to optimize joins in Kafka Streams.
|
| 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.
| ryzhyk wrote:
| The correct way to think about the problem is in terms of
| evaluating joins (or any other queries) over changing datasets.
| And for that you need an engine designed for *incremental*
| processing from the ground up: algorithms, data structures, the
| storage layer, and of course the underlying theory. If you don't
| have such an engine, you're doomed to build layer of hacks, and
| still fail to do it well.
|
| 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.
| d0mine wrote:
| Is it related to Differential Dataflow / timely dataflow
| https://github.com/TimelyDataflow/differential-dataflow
| ryzhyk wrote:
| We have our own formal model called DBSP:
| https://docs.feldera.com/papers
|
| 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.
___________________________________________________________________
(page generated 2024-10-25 23:00 UTC)