Parquet Reboot

[mrocklin]

Our parquet performance is bad. I get 20MB/s in real-world use cases on the cloud where I would expect 500 MB/s. This accounts for ~80% of our runtime in complex dataframe queries in TPC-H. Systems like P2P, the GIL, Pandas copy-on-write, PyArrow strings, etc, are all inconsequential relative to this performance bottleneck.

Unfortunately, improving this situation is difficult because our current Parquet implementation has many layers and many systems all interwoven with each other. What should be a relatively simple system is today somewhat opaque. I would like for us to consider a rewrite.

There are many things to consider here. I'll follow up with some personal thoughts, and I welcome thoughts from others.

[mrocklin]

Personal recommendation

If I were to build this I would remove a lot. I'd remove engines and go Arrow-first with both parquet deserialization and with data access. I think that we could make this a single-python-file effort that had reliably good performance and much easier maintenance.

We would lose some features:

• fastparquet
• index / divisions
• metadata file
• Probably we need to maintain a filesystem= keyword in case someone has an odd filesystem and needs to use fsspec, but this would be the exception I think
• Probably the loss of engines means that RAPIDS folks need to maintain their own read-parquet code paths outside of Dask. I think that this is ok, they've done that before with dask-cudf. Probably they just do it again.

Some things to gain

• There are various settings we should be aware of from [Python][Parquet] Parquet deserialization speeds slower on Linux apache/arrow#38389 that might help improve performance
• Consider keeping RLE columns as categorical
• Preferring LZ4 compression over snappy (lz4 used to not be supported, but maybe is today, and is maybe faster?)
• Use arrow filesystems by default when available

[phofl]

Having a smaller environment because you don’t have to install arrow is the main benefit of fastparquet as far as I am aware of. Pandas will require arrow as a dependency starting with the 3.0 release if there aren’t any unexpected developments. This would render the environment size argument irrelevant

[mrocklin]

There are likely other benefits to fastparquet, but mostly I just want one system rather than having to support many systems and while I like fastparquet, I like pyarrow marginally more.

[fjetter]

+1

I think specializing on pyarrow will be the biggest benefit here that allows to cut complexity.

Additionally, I would like us to review closely how we want to deal with files and rowgroups and the collection of statistics.

I'm very strongly biased to not collect statistics of the entire dataset automatically (if ever) but I appreciate that there are use cases where we have to deal with parquet metadata. I would like us to revisit the existing logic and invest time in reproducers that support the features we are offering and allow us to run tests for them.

Re: single-file-effort. I believe this is something that we also have to review a little more closely. Especially with the introduction of FusedIO in dask-expr there are significant optimizations available when treating multiple files in a batch (pre-fetching IO, for instance). pyarrow implements these capabilities transparently to a certain degree.

[mrocklin]

I'm going to take some time this morning and prototype a very simple implementation. If it works I'll use it for TPC-H. We'll see how far I get. I give it a 50/50 chance that I give up because it's too complex.

[mrocklin]

dask/dask-expr#373

[rjzamora]

Cool - Sorry, I saw dask-contrib/dask-expr#373 but I missed this issue.


I agree that we should simplify things in parquet land, and the best way to do this is to re-write things from scratch. The parquet-engine landscape is dramatically different than it was when the Engine system was first introduced - The PyArrow API and feature set has completely changed since then, and RAPIDS is pretty adaptable on the parquet front anyway.

If you want dd.read_parquet to support hive-partitioned datasets and produce reliably “good” partition sizes, you will end up with a lot of the same bells and whistles that exist now, but you will have the opportunity to organize things in a way that makes sense for the current PyArrow API. Also, now that we have dask-Expr in the mix, the partition-size issue can probably be re-cast as an optimization pass instead of read_parquet arguments.

Somewhat related thoughts: I realize that you are not personally concerned about dask-cudf/rapids - I think that is okay if we can find a way to be smart about “dispatching” Expr classes in Dask-Expr (e.g. some variation or better alternative to dask/dask-expr#321). If downstream projects have some way to replace specific Expr definitions that don’t suit then, then we can probably avoid the kinds of tech debt that the Engine design has caused in the past. One possible outcome is that we use Dask-Expr to start stripping out unnecessary backend/compute dispatching logic in favor of a lower-maintenance coarse-grained dispatching approach.

[mrocklin]

Yeah, I'm hoping that we can defer a lot of the fancy parquet logic to Arrow. ParquetDataset has a lot of stuff like hive partitioning, yes?

Mostly I want to avoid a situation where things get sufficiently complex that people stop maintaining them. This seems to be what happened with parquet. We all knew that it was a trainwreck, but no one wanted to spend the time to make it good.

If there is a plan that makes things easy for others to extend and that doesn't make life harder for folks maintaining the core then I have no objection.

[rjzamora]

I'm hoping that we can defer a lot of the fancy parquet logic to Arrow. ParquetDataset has a lot of stuff like hive partitioning, yes?

Yes, I completely agree that we should defer as much as possible PyArrow. The PyArrow feature set is probably ready to take the reins.

We all knew that it was a trainwreck, but no one wanted to spend the time to make it good.

Yes, it is a mess, but I don't think it is completely a question of time or effort. I'm pretty sure people have just been afraid of breaking the complicated PyArrow/FastParquet/Dask-cuDF/Dask-GeoPandas/etc API contract that was designed at a time when Dask needed to implement most of the ugly dataset parsing/planning logic. We were obviously trying to do something "good" by building a pluggable read_parquet architecture, but the original goal has proven impractical over time.

If there is a plan that makes things easy for others to extend and that doesn't make life harder for folks maintaining the core then I have no objection.

That's good to hear. I'm also pretty motivated to minimize maintenance burden.

Note: It may also be valuable to define simple Expr templates for custom IO Expressions (similar to the way from_map makes it relatively easy to implement custom IO solutions in dask.dataframe). It seems ideal to focus on a simple/performant API and to make it easy for power-users to implement their own custom logic when they need it.

[mrocklin]

Note: It may also be valuable to define simple Expr templates for custom IO Expressions (similar to the way from_map makes it relatively easy to implement custom IO solutions in dask.dataframe). It seems ideal to focus on a simple/performant API and to make it easy for power-users to implement their own custom logic when they need it

As a heads-up, I'll likely be far more nervous / hesitant around reworking things into a complex hierarchy. You should probably expect a lot more friction and demands for design docs ahead of time for anything like this. I'm likely going to push to keep things very flat and simple.

[rjzamora]

Hmm. That was not a suggestion for any kind of hierarchy at all, so not sure I understand. Either way, I hear you.