Pandas + Dask DataFrame 2.0 - Comparison to Spark, DuckDB and Polars

Speaker 1 [00:05]

And welcome to my talk about Pandas and Dask DataFrame 2.0. Pandas 2.0 was like a year ago now, and now we're also ready in Dask DataFrame to unofficially announce like, oh, we're actually pretty fast now. And we'll look at this in the context of how Dask now runs compared to like Spark, which is another distributed tool, and to newer tools like DuckDB and Polars. So, brief intro, I'll keep this short. I work at Coil. Coil is a company that deploys Dask for our customers. It's a managed Dask service. I'm a Pandas and Dask maintainer. I've been a Pandas maintainer since 2021. And I work on Dask now since early last year. Dask, for those of you who haven't come in contact with it yet, is a tool that enables running the PyData ecosystem at a bigger scale than what the tools are normally used for. Today we will focus almost all of the time on our Pandas integration, which provides a data frame on terabytes of data if you want. The talk is kind of split in two different parts, and then we'll look a little bit ahead in the end. We will first investigate what changed in the Dask data frame world and how this reflects usability, performance, scalability. And then we will look ahead into the benchmarking landscape and how Dask now performs compared to Spark, DuckDB, and Polars. And then there are a couple of improvements that are on our immediate roadmap, but more about that a little bit later. The improvements that were enabled by default in Dask over the last 12, 14 months can basically be mostly split into two different categories. One of them is shuffling, strings, and optimization. I'll keep the first two parts brief. These have already been on for quite some time now, and we will focus mostly on our optimizer, which went live six weeks ago, so this is pretty new. The first thing I want to talk about are a more efficient string memory layout compared to what Dask and also Pandas by extension used by default historically. This is PyArrow-backed strings. We enabled them in Dask by default a while ago, and because of mostly two reasons, they are faster, which is nice. Everyone likes it if their stuff runs two times, three times as fast. But they are also, and this is more important for us in the distributed space a lot more memory efficient than NumPy object is. On the right side, we can see a small plot where we just compare memory usage, object detail versus PyEra strings. And this was measured on a data frame that had like two string columns and 20 numeric columns. So even like reducing the two string columns gave us like over 50% improvement. the second part is how we made joints faster so you probably guess like if we need a slide for faster joints than they sucked previously performance wise shuffling is if you have to send your whole data set that's currently somewhere in memory on a distributed cluster over the network. So we will send the data all across our cluster, which is pretty slow. In memory, processing is like 5 to 10 gigabytes per second, and network transfer is like 200 megabytes, not gigabytes. So making this faster has a way bigger impact compared to if we optimize the last 2% out of a Pandas workload, simply because it doesn't really matter much for us. And we had, like, two different things that we could improve with our new shuffle algorithm. It's also based on arrow, so this is kind of, like, the arrow time in Dask now. A data frame normally in Dask has, like, a number of partitions. We call them, like, n here. And previously, we created, like, a lot of, like, tasks that are a scheduler that coordinates the whole workflow I had to handle. and that's scaled with log n times n, which is not good. Especially if you want to scale out. And now the new algorithm, it still has to do all the work that was previously routed through the scheduler, but we were able to reduce the number of tasks to n, so we scale linearly with the number of partitions, which just strains the whole system a lot less. Previously, like Dask, fell over a lot if you scaled out. Now this really isn't a problem anymore. We reduced all this like overhead as was within the system with like low overhead tasks that are now hidden away from the actual system. This is like what you can see here. I don't want to go into detail here, but the middle line is the important thing. Previously like this just blew up based on the number of partitions. And now we are hiding this complexity away with like one single task, which is pretty nice. the previous shuffle algorithm held the data at some point completely in memory which means that you need a cluster that's at least as big as your data set now we write this to disk in the intermediate so we can basically shuffle data at constant memory and what we will see later we can process huge data sets with just a tiny bit of available ram this is our quick like precursor and now we look into how the optimizer which enables us to like do logical query planning which is something that comes out of the database space normally but some of the newer like data frame tools especially polars do this as well and this just makes our life a lot easier like speaking from a user's point of view our optimizer can do like different things i will focus on three or four depends on how you want to categorize them today. They are like a bunch of others that just feed our like task knowledge as maintainers into the queries that we get from our users to replace like very expensive operations with cheap ones. This is how we can save a lot of time but this is like nothing that you can build a talk around. Column projection and predicate pushdown is what we call if you have a data set with like 100 columns and you only need two of them. If you write a dump query like I did here, then in the end the group by operation down here will just collect order total. And all the other like 99 columns that we had in our initial data set are dropped at this stage. Also predicate pushdown is the analogous for the filter for the column projection. This filter is also applied immediately before we finish our query. But there's really no reason to not do this a little bit earlier. The code that we briefly looked at basically translates into this query. We just read data from storage. It doesn't really matter where this is. Then we do an efficient like a very expensive shuffle that just sends all the data over the network. Like, then we'll do the filter, which, if you're lucky, like, reduces our number of rows by, like, quite some amount. And then we do the group by in the end. Like, this is, like, me being pretty dumb, basically. Now, this is what Dask ran, like, two months ago. It just, like, loaded all the data and then shuffled, probably fell on its face along the way before it did the group by right in the end. what we do now is we look at the query and see where we need our columns so we can push like the column restriction down to two columns all the way up to the data source reading so if you have this data like somewhere stored in i don't know s3 or azure blob storage google cloud storage we can just reduce the number of like megabytes that we have to read from from remote storage by quite some amount. This gives you huge improvements if you're not careful while writing your query. Also, there's really no reason that we do the filter after the merge. Doing a filter beforehand won't really impact your merge operation. So we can just move this over. This is a technique that brought us three, four, five times speedups on some queries. It really matters more than one might think coming from, like, a pandas workload where this doesn't matter as much. This technique was, like, general purpose. Like, every tool can and should use that if possible. Now we get a little bit more into techniques that users previously had to do manually and, like, what now the optimizer takes over for them. And also some stuff that, like, requires Dask internal knowledge, how Dask, like, where the bottom legs are and where it performs badly. One of the things is auto-re-partitioning. Historically, Dask looked at your data source, and maybe that's, like, 10,000 different Parquet or CSV or whatever files. And this resulted, then, in, like, 10,000 partitions, which, like, all of them are Pandas data frames. And if we only care about one out of 100 columns, then each and every one of those partitions is very, very small. It's probably tiny, probably like two megabytes or so. And in a distributed setting, it matters that these partitions, if they are too tiny, then you create a lot of overhead for your system. So what we do now is in different steps, we look at how much data of this file do we actually need and then combine multiple into one. This is a short example where we select two out of four columns so we can squash our 10 files into five. But the benchmarks that we'll look at later, there is cases where you only select one out of 20 columns so we can combine 20 files into one partition and just reduce the number of tasks we have to run quite drastically. When we initially added this, this was a five-line code change and it gave us a 2x improvement on the benchmarks that we'll see later, which is, I would say, a good return on investment. The next step is, again, faster merges. Trivial merges is what we call if you don't have to shuffle before you can do a Pandas merge on all your partitions. We have a query here that does two merges right after another. and this results in like a query that looks kind of like this you have a shuffle join here which shuffles left and right and then a dusted join and then you have another shuffle join a little bit further down again shuffles left and right and dusted join which results in like four different shuffles this isn't really necessary because the left side here was already shuffled on the same column so the partition layout wouldn't change We just do a very, very expensive operation for a whole lot of, like, nothing, basically. So what the optimizer does now is it looks at this query and identifies exactly this pattern and then replaces, like, this shuffle join here with a trivial join and just injects another shuffle over here, which means we can, like, reduce the number of shuffles by one, which is, like, gives us probably a 25% improvement one-to-one. This translates into for users, Dask is kind of fast now, we'll see that later, and a lot more reliable. What we also think is that it makes it easier to use for users, this is kind of subjective, so everyone has to come away with his own opinion, but what we can actually quantify is that we scale a lot better, and we're also a lot faster than before. We ran some benchmarks on what Dask looked like one and a half years ago compared to what it looks like now. And what we've seen is that all three of these optimizations that we talked about, like the arrow strings, shuffling, and the optimizer, all gave us pretty significant improvements. Initially, we were at a factor of 19, which translates to 950 seconds, which is kind of slow for a terabyte, down to 50 seconds now. This depends a little bit on the query. Sometimes you might see an even bigger speedup, and sometimes you might see a smaller speedup. Depends how complex your query is, but generally we've seen huge improvements on basically every workload that we run in our benchmarks. Which benchmarking brings us to the second part of the talk. And this means that we look how Dask compares to Spark, DuckDB, and Polars. of the things, this is a TLDR that we come away with, we all suck in some way or another. So there's no clear winner over any scale or anything that we look at. These are the four tools that we look at. There was a keynote earlier today, there's a Polar's talk later, so I won't go super deep in them. Spark is the de facto standard for big data processing at the moment, and Dask is basically in the ETL space like pandas, but big. We have to talk a little bit about the boring part before we can look at some hopefully exciting charts. We are running TPC-H style benchmarks, this is not an accident, we went out there and checked which benchmarks Dask sucked on the most, and ended up with this. Dask is a big data tool, so we won't look at 100 megabyte, 1 gigabyte or so. We start with 100 gigs before we move on to the terabyte space. What's important to know about TPC-H is if we look at a 10 terabyte data set, we won't probably ever touch for any query 10 terabytes of data. They reduce a lot, so in the end you might end up with a single number. But they are queries that reduce more and reduce less and this is something we'll see in the benchmarks later. DuckDB and Polaris are both single machines so they get a single like very very big VM and Dask will get the same number of cores and the same memory same for Spark but many many small machines just that we can utilize like the distributed nature of both frameworks. We'll start with the 100 gigabyte scale and this is actually the only like plot where I put all of the tools in one graph, so it will be a little bit confusing. We have four different tools. Performance for all of them is, I would say, mostly good. DuckDB is actually very, very fast on that scale. So if you have 100 gigs of data and you're sure that this won't grow at some point, then DuckDB is a very, very safe choice. Spark struggles a little bit on this smaller scale it's built for bigger data we'll see that it will perform a bit better later on and Dask, which is the orange bar is actually competitive if you remember, we became 20 times faster over the last year so if you multiply our runtime by 20 then we'll get destroyed here this is what I meant, we really sucked on those benchmarks what I've seen with Polars like initially when I proposed this talk I intended to start with 10 gigs but then we moved on to larger scales Polars is like designed for like medium-sized data so 100 gigs on a 256 gigabyte memory machine is probably a little bit too much at the moment but their actual like compute performance is very very good so if they can like fix a couple of things that they are probably stumbling over for those queries, then they will be a lot faster at some point in the future. But this is mostly the reason why I omit them from 1 terabyte and 10 terabyte, because they tended to time out when we were running the comparisons. So now we look at, for Dask people, the interesting scale, like 1 and 10 terabytes. And we start with Dask versus DuckDB. Again, DuckDB is a single machine node. It's optimized, like, not for really large data because this is something you can't really efficiently process on one machine if you work with petabytes of data. But their performance is actually very, very good. Like, what we see here is we see Dask versus DuckDB. Both have, like, 128 CPUs available. And it's, like, a mixed bag, basically. This is what I meant earlier. Like, we all kind of suck and we're all kind of good. Like, Dask beats DuckDB on half of the queries. DuckDB beats Dask on the other half of the queries. and they struggled, like DuckDB struggled with one query that's like the other tools will also struggle on a larger scale later on. So this is not super surprising. Based on this result, I honestly expected that we would beat DuckDB by a lot on like 10 terabytes, but no. DuckDB struggled a little bit with the queries that don't drop as much data. This is where it ran like out of memory, But it still completed over half of the queries, and it was actually very, very fast in most of the ones that it completed. So on query two, it just kicked Dask's ass by a landslide. We had some queries where we were better. What we'll talk a little bit about later is overhead. There's still a lot to gain for us. We are not really at the limit yet where we can get to with Dask. That's the thing I care about the most, to be honest. Spark is our direct competitor in the distributed space. So we'll start with one terabyte. And to our surprise, we are actually quite fast. We outperformed Spark on most queries. Not by much, just by a little bit. And on average, we were roughly 20% faster. So basically, mostly equal footing, I would say. We struggled a lot with deploying Spark and getting it to run reliable. That wasn't very easy. So it's possible that we missed something that gives us another 20, 30% in Spark performance. The benchmarks are all public, so if someone finds something, we're happy to address that. We're especially happy if someone likes to deploy Spark for us. More interesting was the 10 terabyte case. Initially, like, we just ran DuckDB on, like, 500 gigs of memory. And it completed, like, half of the queries and most of them pretty efficiently. Dask also, like, turned through all the queries mostly happily. When we tried this with Spark, it just wouldn't run. It was working and working and working and working and then one hour timeout hit. Like, runtime here is, like, five minutes, not 60. So it looked like to us that it really struggled with out-of-core performance. So what we did, we increased to 5 terabytes of memory. But we really didn't want to render a machine with 5 terabytes from AWS, so that's why DuckDB isn't in that plot. We outperformed Spark on half the queries that we both had numbers for. And Spark really kicked our ass here in that space. this is like what I previewed a little bit with overhead query 15 is for example a very bad example for us our query runtime is roughly 50 seconds pretty similar to Spark and then we have a lot of overhead until the computation actually starts at the moment these are like kind of known issues that we hopefully address soon and a couple of things are things that popped up, to be honest this conference is like 3 weeks early, I'm hopeful that we are like a lot faster in like 3 weeks to come But what surprised us the most is that, again, we are not Spark experts, but we are reasonably confident that we didn't screw up the deployment majorly. Spark just couldn't run on the queries that processed a lot of data and didn't drop as much. It was running out of memory, timing out, whatever. It was like computing a bunch of stuff and then at some point just fell flat on its face. We looked into other benchmarks where TPC-H with Spark was run on larger scale and to us it looked like that all these things were run with like at least 10 terabytes of memory. So when you could actually fit your whole data set into memory, which is a little bit like not really an out of core mindset. So you would probably get Spark to run through there if you have like a lot more memory available. But this leads you to the risk of like having a huge cluster, even though you only need like 500 gigs of RAM and just like paying huge cloud builds instead of like processing your stuff efficiently. So the main takeaways are like DuckDB actually performs pretty well. It seemed to struggle a little bit on 10 terabyte with the queries that don't drop as much data. That said like DuckDB was just like pip install DuckDB like put in the SQL and then it ran. This was a very different experience to Spark, where running took us quite some time, and we are all task maintainers, so getting tasks run wasn't very hard for us. But that's probably not a fair comparison. Spark on the other side is reasonably fast on larger scale data sets, but it seems to need a lot of resources to actually complete more complex queries that deal with a lot of data. We're happy to be convinced otherwise if we missed something, but I don't really know. And the thing we are happy about is, again, Dask is 20 times faster than one and a half years ago. We are pretty competitive now. If you looked at this one year ago, we would probably have been destroyed by all of these engines because we were just very, very slow. But, and this is where the final slide of my talk comes in, there are still many things that we can actually make faster so that we get a lot faster in total. Again, there's a lot of overhead in some queries. If you can reduce this, then you can probably cut runtime by 30 to 40% of these specific queries. This is, like, not a general purpose thing. This is just something where we can improve, like, half of the queries that we've seen in the benchmarks. I'm more excited about, like, the second part. I initially talked about, like, shuffling is fast now. It's actually still pretty slow. It's just a lot faster than it was before. We have a, like, graph PR that hopefully we'll get in at some point. that reduces the total runtime of most of the queries that actually use a significant amount of shuffling by like 30 to 40%. So it roughly makes shuffle probably three to four times faster if you only look at the shuffle operation. So there's obviously still a lot of room left that we could grow there. Our optimizer, while a lot better than before, is actually still pretty dumb. there are things where we now identify that like a shuffle isn't needed anymore but only if you have a really simple case like fixing the more complex cases is something that's on our immediate roadmap and will hopefully get out soon where we can like probably reduce the number of shuffles and especially like query 18 where Spark, DuckDB and also B struggle a lot, you can probably cut the number of shuffles in half and like push our performance quite more, quite a bit so yeah that's it thanks for listening

Speaker 2 [25:23]

All right, so thank you for the talk. We have a couple of questions. I'll start with the first one. How easy is it to move from a pure Pandas code to Dask and Pandas?

Speaker 1 [25:36]

Hopefully very easy, but pandas is a huge API and a couple of ugly corner cases. But that said, it doesn't hurt to switch the import. Most of our API is exactly the same, it just adds a couple of configurations that are necessary for distributed workload, but it doesn't really hurt trying. The benchmarks we've seen, for example, 99% of the time it's just pandas code that has a different import.

Speaker 2 [26:07]

How does the shuffling in dusk differ from the shuffling in polars?

Speaker 1 [26:12]

Polars really doesn't do shuffling. Polars is on a single machine, so it doesn't have to like bother with network transfer fortunately for them

Speaker 2 [26:25]

Any advantage to use Dask in a single node?

Speaker 1 [26:32]

This depends a little bit on the comparison. It's a lot easier to use Dask on a single node compared to PySpark. It's probably not easier compared to DuckDB or Polars. But what you can do is you can scale out very easily from a single node to a distributed cluster if your Dask code is already running on a single node. So if you need to scale out, there's definitely an advantage to develop locally. And then if you have to process your whole data set, just use whatever resources you need.

Speaker 2 [27:02]

If Dask does optimizations for me is it possible for me as a user to accidentally do something where Dask will now run slower than before?

Speaker 1 [27:13]

No, don't get me wrong, you can do dumb things that will actually block all optimizations, but it's just the same shit that you had before.

Speaker 2 [27:27]

This question, I believe you answered this during your talk, but since it has upwards, I will ask. What does this middle job on the P2P shuffling algorithm do which previously blew up to the O and login?

Speaker 1 [27:40]

Right now we are hiding all the complexity away and the workers are communicating with each other instead of like sending everything through the scheduler. So the middle job is basically just please wait until like the transfer is finished so that I can start actually using the data that was sent over the network. It doesn't really do anything it's just like a little bit of like a sleep.

Speaker 2 [28:03]

Have you compared Mojin with Dask? If so, what's faster?

Speaker 1 [28:08]

No, we haven't. Modin is fast in some cases from what I've seen and really, really slow in others, but I can't really give any more details than that.

Speaker 2 [28:21]

I think this would be the last question. How much overhead is there from optimization in Dask and are there cases when it's not worth it?

Speaker 1 [28:30]

If your computation runs in probably like 500 milliseconds, then the first thing I can tell you is like you shouldn't be using task anyway then the overhead is like Noticeable, but if you have a like query that runs for like 20 to 30 seconds, then it doesn't really matter

Speaker 2 [28:49]

Thank you for your questions, and I apologize for the ones that we couldn't get to. Can people find you afterwards?

Speaker 1 [28:56]

afterwards. I'll be around and I'll also be at the Coiled booth if anyone wants to chat.