Pyarrow 8.0.0 write_dataset writes data in different order with use_threads=True

[asfimport]

In the latest (8.0.0) release the following code snippet seems to write out data in a different order for each of the partitions when use_threads=True vs when {}use_threads=False{}.

Testing the same snippet with pyarrow 7.0.0 gives the same order regardless of whether use_threads is set to True when the data is written.

 

import itertools

import numpy as np
import pyarrow.dataset as ds
import pyarrow as pa

n_rows, n_cols = 100_000, 20

def create_dataframe(color, year):
    arr = np.random.randn(n_rows, n_cols)
    df = pd.DataFrame(data=arr, columns=[f"column_{i}" for i in range(n_cols)])
    df["color"] = color
    df["year"] = year
    df["id"] = np.arange(len(df))
    return df


partitions = ["red", "green", "blue"]
years = [2011, 2012, 2013]
dataframes = [create_dataframe(p, y) for p, y in itertools.product(partitions, years)]
df = pd.concat(dataframes)

table = pa.Table.from_pandas(df=df)

ds.write_dataset(
    table,
    "./test",
    format="parquet",
    max_rows_per_group=1_000_000,
    min_rows_per_group=1_000_000,
    existing_data_behavior="overwrite_or_ignore",
    partitioning=ds.partitioning(pa.schema([
        ("color", pa.string()),
        ("year", pa.int64())
    ]), flavor="hive"),
    use_threads=True,
)

df_read = pd.read_parquet("./test/color=blue/year=2012")
df_read.head()[["id"]]

 

Tested on Ubuntu 20.04 with Python 3.8 and arrow versions 8.0.0 and 7.0.0.

Reporter: Daniel Friar

Related issues:

• [C++][Dataset] Preserve order when writing dataset (duplicates)

_{Note: This issue was originally created as ARROW-16506. Please see the migration documentation for further details.}

[asfimport]

Weston Pace / @westonpace:
This is expected behavior at the moment. I'm guessing at some point between 7.0.0 and 8.0.0 we switched to reading from a table in a multi-threaded fashion.

Is an ordered write important? There has been some discussion of adding sequencing on the mailing list. If we have sequencing information on batches then we could use that to do an ordered write.

[asfimport]

Daniel Friar:
Thanks @westonpace, that makes sense. I wonder if it's worth adding to the docs just to make it clear that this is the case and ordered writes should not be expected/relied upon?

I do think there are cases where an ordered write is important, particularly for e.g. time series data where a particular ordering is necessary for downstream tasks and additional in-memory sorts once the data is loaded may be expensive.

More of an opinion but generally I think it can be a little surprising to write from a table/dataframe and read back to discover the order has changed. Adding sequencing to achieve an ordered write would be an improvement IMO!

[asfimport]

Weston Pace / @westonpace:
FYI ARROW-16518 was recently filed and I was reminded of ARROW-10883. I am going to mark this as a duplicate of ARROW-10883.

Adding a section on ordering to the docs is a good idea. Yes, there was a recent discussion on the ML regarding an as-of join (i.e. something you commonly fined in a time series database) and sequencing was a part of that discussion.