cc @jorisvandenbossche and @martindurant , who might be interested in this

The file_path in the pieces themselves should not be populated, though, because they ought to be readable as individual files if need be, but the metadata chunk passed back should contain it. Is my thinking right?

The file_path in the pieces themselves should not be populated, though, because they ought to be readable as individual files if need be, but the metadata chunk passed back should contain it. Is my thinking right?

Yes - We want the "file_path" field in "_metadata" to be correctly populated (with a relative path to the root directory where "_metadata" is). However, the footer metadata in the actual parquet-data files should not have this field populated.

Note that the latest changes allow the ArrowEngine to generate "correct" "_metadata" files for partitioned datasets. However, this required me to do the actual partitioning within dask (only a bit of code needed to be borrowed from pyarrow). Since the file_path metadata field is now populated, we can sort the row_groups by the path in the same way that we sort the pieces (which ensures that the filtering is correct).

Any thoughts here @martindurant ?

I am happy to look again, but I thought you were waiting for clarity from pyarrow?

I am happy to look again, but I thought you were waiting for clarity from pyarrow?

Right, I was hoping to get thoughts from @jorisvandenbossche about the possibility of avoiding changes in dask in cases where the changes could be made in pyarrow. However, I would really like to support proper "_metadata" creation/handling for current/older version of pyarrow. So, even if we can get the necessary changes into pyarrow, I do think we need to temporarily move the partitioning logic from write_to_dataset into dask.

Taking a look now.

But indeed, even if there are some changes that could be made to pyarrow's write_to_dataset, you probably want to include them here already to avoid needing to wait on pyarrow (if it doesn't add too much code)

I actually didn't realise dask was using write_to_dataset (as I was thinking that since each partition of the dask DataFrame needs to be written separately (and to a separate file(s)), you need to handle multiple writes anyway. But I suppose a partition key is not necessarily unique within one dask DataFrame partition, so in that way you still make use of the groupby-functionality of write_to_dataset ?

No, partition keys are not unique within partitions. You end up with files like

/key=1/part.0.parq
/key=1/part.1.parq
/key=2/part.0.parq
/key=2/part.1.parq

where the part.0s are from partition 0, etc.

About changing the path, that might depend on whether there are other use cases for collecting this metadata (although I assume dask is the main user of this keyword).

A github search turned up dask, cudf and spatialpandas as users of the metadata_collector keyword. I assume cudf needs the same fix as dask. I didn't check yet how it's used in spatialpandas.

One more thing I am wondering: what happens if you specify the combination of split_row_groups=False, gather_statistics=True ? Is filtering supported then?

This is a great question. Originally, this was not supported (at least the result would be wrong). I just added some logic to merge row-group statistics for each file -- Hopefully this is supported "correctly" now.

Does this impact #6017 ?

Does this impact #6017 ?

These changes are entirely focused on ArrowEngine - I believe 6017 should only affect FastParquetEngine

Is row-group-level filtering (multiple row-groups within a file) useful at all? This all seems to be leading towards real row-level filter pushdown...

I am not much of a "user," so it is hard for me to say (with confidence) that it is useful. However, I have seen it used before in practice a few times (people dealing with multiple row-groups in very large parquet files).

Regardless of how useful it is, the row-group definitely does seem like it should be the smallest "unit" of filtering to me. Whether or not there are multiple row-groups in a file should not affect whether or not filtering is properly supported.

The comment on row-filtering is only pie-in-the-sky, no action for us. Indeed, the filtering would have to happen in the arrow internals anyway.

@TomAugspurger , could you have a look at the unrelated pandas csv failure?

@martindurant that looks like #5910 and can probably be ignored.

I have NEVER seen multiple row groups in a file where the dataset as a whole is partitioned.

I don't know how common it is in practice, but this is actually the case how this all started. I had written such a file (by specifying a row_group_size), which lead me to report the buggy behaviour in #5959.

OK, I am happy with everything here.

OK, I am happy with everything here.

I really appreciate your time on this @martindurant ! It was also awesome to have your eyes on this @jorisvandenbossche !

Hey @rjzamora,
Sorry for reopening this particular change. It took me a while to find the commit that removed the write_to_dataset function from pyarrow within the dask code.
I am currently facing the following issue:

I'm working with time-series data, which I want to save in a date-partitioned format. Every day I calculate some features a few days ahead and want to save the new calculations in my partitioned data lake in a way that overwrites matching partitions but keeps old partitions.
Pyarrow has a nice functionality for this: pyarrow.parquet.write_to_dataset(existing_data_behavior='delete_matching')
Testing all kinds of combinations of the append, overwrite and ignore_divisions arguments of dask, I was never able to reproduce the wanted behavior.

Following the documentation of dask I read about the different engines ('fastparquet', 'parquet' and earlier also 'parquet-dataset') and the possibility to pass **kwargs down to the specific engine.
Now 'parquet-dataset' was removed as an engine, but seems to be the default, because you are using the ArrowDatasetEngine class.

So with this information I assumed that I could pass the following kwarg to the dask function: dd.to_parquet(existing_data_behavior='delete_matching')
Well turns out I was wrong, because it doesn't recognise the keyword 'existing_data_behavior'.
Turns out the ArrowDatasetEngine class has a very misleading name and the pyarrow engine used by dask is actually the legacy pa.write_table() function.

So after this introduction and deep dive into dask, here come my questions:

1. Are you guys aware that you are actually not making use of the newer pyarrow.dataset engine when it comes to actually writing the dataset to parquet files? (You are only using it for the initialize_write() function.)
2. Is there currently a way for me to replicate the wanted behavior of 'delete_matching' with the available arguments within dask? (maybe with fastparquet? I did not yet deep-dive into that part, because I do prefer using pyarrow)
3. If not, are there any plans to actually step over to the newer pyarrow.dataset engine in order to make fully use of pyarrows capabilities?

Kind regards,
Kilian

P.S.: Sorry for not creating a new issue with this. I thought it might be helpful to link the change that removed the pyarrow.dataset engine from the dask code base to my question.

1. Are you guys aware that you are actually not making use of the newer pyarrow.dataset engine when it comes to actually writing the dataset to parquet files? (You are only using it for the initialize_write() function.)

Yes - The write API was a pretty late addition to pyarrow.dataset, and is a bit less convenient than the read API for a parallel framework like Dask. To say this another way: The "arrow-dataset" engine was originally added at a time when pyarrow.dataset could only be used to read existing data, and Dask's ArrowDatasetEngine has not been updated to actually use these newer write features yet.

Since the Dask programming model more-or-less assumes that we can/will write out each partition in an embarrassingly-parallel fashion, using a feature like existing_data_behavior='delete_matching' is far from simple. In order to use this feature "safely" we would need to add a lot of validation logic or completely change the to_parquet algorithm to guarantee that we are not overwriting the wrong data (e.g. we need to know that concurrent processes i and j cannot be writing data to the same directory partition).

2. Is there currently a way for me to replicate the wanted behavior of 'delete_matching' with the available arguments within dask? (maybe with fastparquet? I did not yet deep-dive into that part, because I do prefer using pyarrow)

If your data is already partitioned (in the Dask-DataFrame sense) in such a way that you know data from two distinct Dask partitions will never be written to the same parquet-directory partition, then you may be best-off implementing this logic yourself. For example, I haven't tested this code myself, but I can imagine you doing something like the following:

ddf = <your collection>
base_dir = <root of your partitioned dataset>

# Define custom logic to write a partition to disk
def custom_write(df):

    # Convert df to an arrow table, and then use pyarrow's dataset API as you wish... 
    ds.write_dataset(table, base_dir , …, existing_data_behavior=‘delete_matching’)

    # Probably simplest to return an empty dataframe to make map_partitions happy
    return ddf._meta

# Call/compute map_partitions to write out your data
# using your own custom logic
ddf.map_partitions(custom_write).compute()

If not, are there any plans to actually step over to the newer pyarrow.dataset engine in order to make fully use of pyarrows capabilities?

I'm not aware of any explicit plans, but I'd be open to it if such an update would be beneficial.

Thank you for your quick reply!
Okay, that makes it more understandable why you were referring to it as 'pyarrow-dataset' engine.

I'm new to dask and the entire split-up-dataframe mechanics, but I think I understand enough to see that this behavior might be very complex to implement without running the risk of deleting other dask-dataframe partitions.

About my dataset: I know that it is already sorted by date, but I don't know if the underlying partitions are split in between dates or if the same date can exist in multiple partitions. How can I ensure that the second one is not the case?
Because I have multiple rows associated with the same date. I guess that's also why I cannot use date as an index within dask, because it is not unique per row.

And in general with regards to your last sentence, I think it's always beneficial to stay up-to-date with the latest features and changes of the underlying packages. But I can imagine that it would take quite some effort to implement those changes, especially since writing to parquet is just one of the many functionalities of dask.

@rjzamora I tried with just deleting the existing partitions above my minimum date myself. But now I get an error when trying to load the dataset again, because of course the metadata file is not updated with the removed partitions.
So I guess manually deleting partitions is not a good approach to this? Or is there an easy way to also remove those partitions from the metadata file?