Enable multi-file partitioning in dask_cudf.read_parquet

[rjzamora]

Depends on dask#7557

This PR updates dask_cudf.read_parquet to enable multi-file aggregation when the chunksize parameter is used. This change enables dramatic (>50x) improvements when the dataset contains many small files, which is somewhat common for hive/directory-partitioned data.

Motivating Example

dask.datasets.timeseries(
    start='2000-01-01',
    end='2000-02-28',
    freq='1s',
    partition_freq='1h',
    seed=42,
).to_parquet(
    path,
    engine="pyarrow",
    partition_on="name",
    write_index=False,
)

####
#### BEFORE THIS PR ####
####

# Without useing `chunksize`, or  when using `chunksize` before this PR
# results in 36,192 partitions
%time df_read = dask_cudf.read_parquet(path)
CPU times: user 1.17 s, sys: 501 ms, total: 1.67 s
Wall time: 1.67 s

# Without useing `chunksize`, or  when using `chunksize` before this PR
# results in a 5+ minute read time
%time df_read.compute()
CPU times: user 5min 34s, sys: 37.1 s, total: 6min 11s
Wall time: 6min 12s

####
#### AFTER THIS PR ####
####

# Using `chunksize` WITH this PR results in 26 partitions
%time df_read = dask_cudf.read_parquet(path, chunksize="1GB", aggregate_files="name")
CPU times: user 1.03 s, sys: 119 ms, total: 1.15 s
Wall time: 1.15 s

# Using `chunksize` WITH this PR results in a ~95x speedup
%time df_read.compute()
CPU times: user 3.18 s, sys: 703 ms, total: 3.88 s
Wall time: 3.92 s

[codecov]

Codecov Report

❗ No coverage uploaded for pull request base (branch-21.08@67b7aac). Click here to learn what that means.
The diff coverage is n/a.

❗ Current head caf2b90 differs from pull request most recent head 08d12e2. Consider uploading reports for the commit 08d12e2 to get more accurate results

@@               Coverage Diff               @@
##             branch-21.08    #8393   +/-   ##
===============================================
  Coverage                ?   10.60%           
===============================================
  Files                   ?      116           
  Lines                   ?    18606           
  Branches                ?        0           
===============================================
  Hits                    ?     1974           
  Misses                  ?    16632           
  Partials                ?        0           

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 67b7aac...08d12e2. Read the comment docs.

[rjzamora]

@randerzander - dask#7557 was merged, and I just confirmed that this PR still enables (very performant) multi-file aggregation via the aggregate_files parameter in read_parquet :)

[ajschmidt8]

Removing ops-codeowners from the required reviews since it doesn't seem there are any file changes that we're responsible for. Feel free to add us back if necessary.

[devavret]

No need for tests?

[rjzamora]

No need for tests?

Good question actually - If CI is using the Dask main branch, we can add an explicit test with aggregate_files= set. Otherwise, we cannot really test for the feature we are enabling. Perhaps a simple test with a version check is appropriate.

[robertmaynard]

Removing cmake-codeowners from the reviews since it doesn't seem this has any build-system impact.

[rjzamora]

@devavret - Just added test coverage

[rgsl888prabhu]

Should this be ?

Suggested change

	pieces = [pieces]
	pieces = list(pieces)

in case pieces is a numpy array or tuple

[rjzamora]

Sorry - I know this is a bit ugly :/

We cant use list(pieces), because pieces will either be a tuple, a string, a list of strings, or a list of tuples. For newer versions of Dask, this should always be a list already. For older versions, however, we want to convert a string into a list of strings, and a tuple into a list of tuples.

[rgsl888prabhu]

Where is this being used ?

[rjzamora]

It is used upstream (in Dask) to check if we support pieces being passed in a as a list in read_partition. So, it is an "opt-in" mechanism.

[quasiben]

Thanks @rjzamora . This looks good and thanks for adding the tests.

Also, thank you to @rgsl888prabhu and @devavret for the review

[quasiben]

@gpucibot merge