Another advantage to push is that it can be pretty straightforwardly distributed—just replace the "push" mechanism with a queue connected to a network stack, and now your operators can run on different machines.
In a batch context this is most useful for sending computation to where the data is, but it's also the foundation for modern stream processing systems (like Flink or Arroyo) which are built over distributed dataflow DAGs.
Interesting. I have the opposite intuition. With a push-based model you end up needing some sort of back-pressure mechanism which gets complicated in a hurry. With a pull-based model you get back pressure "for free"
Yeah, you definitely need backpressure, and doing that well is had. There's a great blog post about how Flink manages this [0] and it involves a complex credit-based flow control system.
But in practice the performance advantages of allowing operators to do work as they have capacity (rather than being potentially starved by the capacity of their downstreams) outweighs this.
I worked on a push-based streaming system which could arbitrarily distribute each operate to another compute node. Backpressure occurred naturally: if the thing you’re pushing into is busy (internal queue is full; TCP socket is busy) you wait. We had no explicit backpressure mechanism, and yet we had it in our system.
This is an interesting point. In Arroyo [0] (the streaming engine I work on) we also use a simple TCP-based backpressure mechanism: we have NxM TCP connections between each operator subtask, behind small in-memory queues. This essentially is relying on the kernel's flow control and backpressure mechanisms instead of building a custom network stack like Flink does.
This has worked well in practice, but I think does leave some performance on the table and likely incurrs higher resource utilization. Flink for example uses a system of "floating" buffers that can be moved between streams as needed to minimize the total number of buffers needed.
However it may be that Linux's network stack has gotten good enough at this point that it makes sense to just rely on it directly.
I wish there was more research out there on how these sorts of network stacks should be built on modern hardware and OSes.
(semi-relatedly, was checking out your CV; a bunch of great looking papers there that I'll definitely be checking out!)
Interesting - I looked into your code a bit. I found your window aggregation library [1]. You may be interested in looking into the Rust implementation of some of the research work I've been a part of [2].
In Flink, I believe the reason they need to implement their own backpressure system is that they multiplex TCP connections. That is, they have multiple logical streams flowing through a single TCP connection. If that's the case, you need to do some work to 1) detect which logical stream is the one that's blocking, and 2) don't block because other logical streams may be able to use the active TCP connection.
Thinking it through, I think what Flink's approach buys is not necessarily better performance, but better just a manageable number of connections. That is, imagine you have a process P1 with operators A, B and C. And then P2 has D, E, F. Now imagine that this is a shuffle, where A, B and C are fully connected to D, E and F. In my old system, you would have 9 TCP connections. In Flink, you will have 1.
Thanks for that link! We've been heavily inspired by the work out of TU Berlin on Scotty (https://tu-berlin-dima.github.io/scotty-window-processor/) but exciting to see more work on this problem. Our efficient window handling is a big area of improvement over what Flink is able to do right now.
For Yep, Flink has to do this themselves because they multiplex. My sense is that when Flink was originally designed in ~2012 Linux didn't support large numbers of TCP connections well (and memory was more of an issue), so they built a network stack that multiplexes many logical streams on top of a single TCP connection.
My (not terribly informed) opinion is that TCP behaviors and semantics are not necessarily optimal for these systems and you get benefits from allowing your systems' scheduler to control prioritization rather than handing it off the kernel.
Yes, multiplexing data flows that need flow control on top of protocols that have flow control is a difficult problem.
SSHv1 had multiplexing but no flow control in the SSHv1 protocol so if you did a bulk transfer over a forwarded port your interactive session channel could "freeze" if the bulk transfer got flow controlled on the receiving end or if it took up all available bandwidth.
SSHv2 fixed this by adding flow control at the SSHv2 channel layer protocol, but the price paid for this was that that flow control essentially acts as a brake on each channel's bandwidth, so you just can't use SSHv2 for high-bandwidth bulk transfers unless you have implementations tuned for that (meaning that they set huge channel windows).
The analogy of a library vs a bookstore to explain the distinction between push and pull query engines is quite effective. It really simplifies a complex topic and make it more relatable.
12 comments
[ 160 ms ] story [ 2998 ms ] threadDiscussion at the time (35 comments): https://news.ycombinator.com/item?id=27006476
It allows decoupling of the control flow and the data flow. The movement of the data in the query pipeline is controlled explicitly.
We did this migration a few years ago. Many database engines forked or influenced by ClickHouse still use pull-based query engines.
In a batch context this is most useful for sending computation to where the data is, but it's also the foundation for modern stream processing systems (like Flink or Arroyo) which are built over distributed dataflow DAGs.
But in practice the performance advantages of allowing operators to do work as they have capacity (rather than being potentially starved by the capacity of their downstreams) outweighs this.
[0] https://flink.apache.org/2019/06/05/a-deep-dive-into-flinks-...
This has worked well in practice, but I think does leave some performance on the table and likely incurrs higher resource utilization. Flink for example uses a system of "floating" buffers that can be moved between streams as needed to minimize the total number of buffers needed.
However it may be that Linux's network stack has gotten good enough at this point that it makes sense to just rely on it directly.
I wish there was more research out there on how these sorts of network stacks should be built on modern hardware and OSes.
(semi-relatedly, was checking out your CV; a bunch of great looking papers there that I'll definitely be checking out!)
[0] https://github.com/ArroyoSystems/arroyo
In Flink, I believe the reason they need to implement their own backpressure system is that they multiplex TCP connections. That is, they have multiple logical streams flowing through a single TCP connection. If that's the case, you need to do some work to 1) detect which logical stream is the one that's blocking, and 2) don't block because other logical streams may be able to use the active TCP connection.
Thinking it through, I think what Flink's approach buys is not necessarily better performance, but better just a manageable number of connections. That is, imagine you have a process P1 with operators A, B and C. And then P2 has D, E, F. Now imagine that this is a shuffle, where A, B and C are fully connected to D, E and F. In my old system, you would have 9 TCP connections. In Flink, you will have 1.
[1] https://github.com/ArroyoSystems/arroyo/blob/master/arroyo-w... [2] https://github.com/IBM/sliding-window-aggregators/tree/maste...
For Yep, Flink has to do this themselves because they multiplex. My sense is that when Flink was originally designed in ~2012 Linux didn't support large numbers of TCP connections well (and memory was more of an issue), so they built a network stack that multiplexes many logical streams on top of a single TCP connection.
My (not terribly informed) opinion is that TCP behaviors and semantics are not necessarily optimal for these systems and you get benefits from allowing your systems' scheduler to control prioritization rather than handing it off the kernel.
There are also a new class of datacenter-oriented protocols like Homa (https://homa-transport.atlassian.net/wiki/spaces/HOMA/overvi...) that I think are pretty interesting to solve this class of problems.
SSHv1 had multiplexing but no flow control in the SSHv1 protocol so if you did a bulk transfer over a forwarded port your interactive session channel could "freeze" if the bulk transfer got flow controlled on the receiving end or if it took up all available bandwidth.
SSHv2 fixed this by adding flow control at the SSHv2 channel layer protocol, but the price paid for this was that that flow control essentially acts as a brake on each channel's bandwidth, so you just can't use SSHv2 for high-bandwidth bulk transfers unless you have implementations tuned for that (meaning that they set huge channel windows).
Nesting flow control just doesn't work very well.