Add new dask_cudf.read_parquet API
#17250
Conversation
rjzamora
added
improvement
rjzamora
changed the title
dask_cudf.read_parquet APIdask_cudf.read_parquet API
rjzamora
added
3 - Ready for Review
| elif ( | ||
| filters is None | ||
| and isinstance(dataset_kwargs, dict) | ||
| and dataset_kwargs.get("partitioning") is None | ||
| ): | ||
| # Skip dataset processing if we have no filters | ||
| # or hive/directory partitioning to deal with. | ||
| return paths, row_groups, [], {} |
There was a problem hiding this comment.
The pyarrow.dataset logic below has non-negligible overhead on remove storage. This code block allows us to pass in dataset_kwargs={"partitioning": None} to skip unnecessary PyArrow processing when we know we are not reading from hive-partitioned data (and we aren't applying filters).
By default (when we don't pass in dataset_kwargs={"partitioning": None}), cudf will still pre-process the dataset with PyArrow just in case it is hive partitioned.
There was a problem hiding this comment.
Thanks for the context @rjzamora ! Do you happen to have the rough overhead numbers/magnitude?
There was a problem hiding this comment.
The overhead is measurable but not huge (~4-5%) for the specific case I was benchmarking.
| **to_pandas_kwargs, | ||
| ) | ||
| class CudfReadParquetFSSpec(ReadParquetFSSpec): | ||
| _STATS_CACHE: MutableMapping[str, Any] = {} |
There was a problem hiding this comment.
nit: Should this be a cache of bounded size (via lru_cache?
There was a problem hiding this comment.
I don't think I'm very worried about this cache getting too large. We're just storing a dict with two elements (schema name, and storage size) for each column.
rjzamora
added
5 - Ready to Merge
|
/merge |
|
@GregoryKimball @vyasr - FYI: I'd really like this to be included in 24.12 (it makes it much easier to use/demonstrate the recent KvikIO + S3 improvements). |
|
All good! I had this on the board slated for this release. |
#17250 started using `pynvml` but did not add the proper dependency, this change fixes the missing dependency. Authors: - Peter Andreas Entschev (https://github.com/pentschev) - https://github.com/jakirkham Approvers: - GALI PREM SAGAR (https://github.com/galipremsagar) - https://github.com/jakirkham URL: #17386
Description
It's time to clean up the
dask_cudf.read_parquetAPI and prioritize GPU-specific optimizations. To this end, it makes sense to expose our ownread_parquetAPI within Dask cuDF.Notes:
dask_cudf.read_parquetAPI is only relevant when query-planning is enabled (the default).filesystem="arrow"now usescudf.read_parquetwhen reading from local storage (rather than PyArrow).blocksizeargument is now specific to the "smallest" NVIDIA device detected within the active dask cluster (or the first device visible to the the client). More specifically, we usepynvmlto find this representative device size, and we setblocksizeto be 1/32 this size.blocksize=0.125to use1/8the minimum device size (orblocksize='1GiB'to bypass the default logic altogether).blocksizeisNone, we disable partition fusion at optimization time.blocksizeis notNone, we use the parquet metadata from the first few files to inform partition fusion at optimization time (instead of a rough column-count ratio).Checklist