I have a use case where I want to store multiple Dask dataframes into a common parquet storage through to_parquet(ddf, 'TestParquet', append=True).
The structure of the parquet files is set through the first dataframe being written to it (without append=True).
Each dataframe has Categorical Index column.
The categories are known from the beginning across all dataframes, and no dataframe have categories in common.
The dataframes are partitioned on the entire list of categories (thus each have empty partitions before getting saved to parquet). Ultimately, once all daframes will be saved to parquet, all categories/partitions will contain data.
The problem: After appending the second dataframe, the index cannot be used to retrieve specific categories.
Here is a minimal reproducible example:
pandas: 0.24.2 fastparquet:0.4.1 dask: 2.22.0
import pandas as pd
import dask.dataframe as dd
import fastparquet
data= {'Name': [ 'C', 'B','F', 'B', 'F', 'B'], 'ID':[1, 1, 2, 2, 3, 3], 'Value':[2,3,4,1,2,3]}
df = pd.DataFrame(data)
df['Name']=df['Name'].astype(pd.api.types.CategoricalDtype(categories=['A', 'B', 'C', 'D', 'E', 'F'], ordered=True) )
ddf = dd.from_pandas(df, npartitions=2)
ddf=ddf.set_index('Name', sorted=False).repartition(divisions=['A', 'B', 'C', 'D', 'E', 'F', 'F'], force=True)
ddf_parquet=dd.read_parquet('./TestParquet')
data2= {'Name': ['D', 'E', 'A', 'A','D', 'A'], 'ID':[1, 1, 2, 3, 3, 4], 'Value':[1,2,3, 4,5,6]}
df2 = pd.DataFrame(data2)
df2['Name']=df2['Name'].astype(pd.api.types.CategoricalDtype(categories=['A', 'B', 'C', 'D', 'E', 'F'], ordered=True) )
ddf2 = dd.from_pandas(df2, npartitions=2)
ddf2=ddf2.set_index('Name', sorted=False).repartition(divisions=['A', 'B', 'C', 'D', 'E', 'F', 'F'], force=True)
dd.to_parquet(ddf2, './TestParquet', engine='fastparquet', append=True, ignore_divisions=True)
ddf_parquet2=dd.read_parquet('./TestParquet')
After saving the first Dataframe to parquet, I can use the index without issue:
ddf_parquet.loc['B'].head()
ID Value
Name
F 3 2
F 2 4
However, after appending the second dataframe, trying to select anything but the index value of the first partition ('B') results in an error:
ddf_parquet2.loc['A'].head()
Traceback (most recent call last):
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4805, in get_slice_bound
return self._searchsorted_monotonic(label, side)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4756, in _searchsorted_monotonic
return self.searchsorted(label, side=side)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/base.py", line 1501, in searchsorted
return self._values.searchsorted(value, side=side, sorter=sorter)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/arrays/categorical.py", line 1370, in searchsorted
raise ValueError("Categorical not ordered\nyou can use "
ValueError: Categorical not ordered
you can use .as_ordered() to change the Categorical to an ordered one
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/dask/dataframe/methods.py", line 42, in try_loc
return loc(df, iindexer, cindexer)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/dask/dataframe/methods.py", line 28, in loc
return df.loc[iindexer]
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexing.py", line 1500, in __getitem__
return self._getitem_axis(maybe_callable, axis=axis)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexing.py", line 1867, in _getitem_axis
return self._get_slice_axis(key, axis=axis)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexing.py", line 1533, in _get_slice_axis
slice_obj.step, kind=self.name)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4673, in slice_indexer
kind=kind)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4872, in slice_locs
start_slice = self.get_slice_bound(start, 'left', kind)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4808, in get_slice_bound
raise err
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4802, in get_slice_bound
slc = self._get_loc_only_exact_matches(label)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 4772, in _get_loc_only_exact_matches
return self.get_loc(key)
File "/cba/local/mx/Linux_x86_64/python/Python-3.6.2/lib/python3.6/site-packages/pandas/core/indexes/category.py", line 438, in get_loc
raise KeyError(key)
KeyError: 'A'
I have no clue, why the 'A' key is not found in the index and why the error suggest to order the categorical as the 'Name' categories were ordered before setting the index.
One observation:
while the divisions for both dataframes are set as expected as :
('A', 'B', 'C', 'D', 'E', 'F', 'F')
After saving to Parquet and reading back the data, the divisions are realigned to the non-null partitions of the first Dataframe after saving the first Dataframe:
('B', 'C', 'F', 'F')
Consequently I have to append the second Daframe with ignore_partitions=True else I am getting an error stating the divisions are overlapping (which is precisely why I repartitioned on the entire list categories prior to saving in an attempt to have the final partitions/divisions from the very first dataframe).
Resetting the index and re-indexing is not really viable as the actual dataset is massive (circa 200GB in total before saving to Parquet).
Setting the divisions of each dataframe to match its actual categories, results in the same error in the end.
Any help would be much appreciated.