fjetter
commented
4 years ago I believe most of the kartothek.serialization module should be handled by pyarrow.
What we do there is restricted to single file actions, e.g. apply predicates, filter row groups, buffer storage access, etc. We built this at a time where pyarrow didn't support anything like this. I believe we could already get rid of a huge chunk of code if we would migrate to the filters of pa.read_pandas. We would probably discover a lot of issues regarding type stability since we invested a lot into this already but I guess both projects would benefit from this.
Moving on to the actual datasets...
I believe we would benefit the most if there was an interface like
import pyarrow.dataset as ds
part_spec = KartothekPartitionSpec() # This doesn't exist but think of it as a replacement to the filesystem crawling
dataset = ds.dataset(part_spec, format="parquet", ...)
query_plan = dataset.to_table(columns=[...], filter=...)
# The interface should somehow return a pointer to the data and the local filters which need to be applied to this shard.
splits == [
("path_to_file", local_filters, ...), # potentially also with more statistics for query optimisation, if available
...
]If there was a way we can interface and you would perform the partition pruning, etc. this would be helpful. For a kartothek integration this would require the concept of secondary indices (which we love dearly), which is a very big topic if you want to define some sort of standard. Our kinds of indices are very simple, though, and we wouldn't need anything sophisticated.
Another part of this is, of course, pre-aggregated or pre-collected file statistics which may or may not be included here. This is something kartothek doesn't do at the moment but I can see this being added in the future.
The output, which I referred to as "splits" for the lack of a better word (please educate me if there is a commonly used term for this data) is where it becomes interesting from my POV. This can be used to construct a dask graph, optimise the query, perform joins with other datasets, etc. I'm wondering what the ambition of arrow is for the tasks/splits. imho, the biggest downside of the arrow dataset is that it cannot work on larger than memory data but this is also why we built kartothek (one of the reasons)
jorisvandenbossche
xhochy
In the Apache Arrow C++ project, we have been working the last moths on a Dataset API (original design document: https://docs.google.com/document/d/1bVhzifD38qDypnSjtf8exvpP3sSB5x_Kw9m-n66FB2c/edit). A first iteration with python bindings is available in pyarrow 0.16 (but it's not yet considered stable).
The dataset API has the goal to handle (scan/materialize) data sources larger than memory, and specifically provide:
Think: the current pyarrow.parquet.ParquetDataset functionality, but then not specific to parquet, and not tied to python (so for example the R bindings of arrow can also use this).
I suppose the second item is not of much interest for kartothek, but the first and third might be.
A small illustrative example of how it can look like in python:
(this uses the discovery mechanisms of Arrow, but you can also construct a Dataset manually with a list of files and partition expressions, if those are already known)
The reference docs: https://arrow.apache.org/docs/python/api/dataset.html (but prose docs with examples is still lacking for now)
Long term, we would like to have this replace the python implementation of
pyarrow.parquet.ParquetDataset. Now, kartothek doesn't actually useParquetDataset, but only basicpyarrow.parquet.read/write_tableandpyarrow.parquet.ParquetFile(to get metadata, statistics, read row group, etc of a single file). And those APIs are normally there to stay (regardless of the datasets API work).However, kartothek still implements similar things (eg row group filtering with predicate pushdown) that would also be covered by Arrow Datasets. So although there is no immediate action required for kartothek due to those changes in Arrow (eg to replace deprecated APIs), I think it would still be useful to think about to what extent kartothek could leverage this work being done in Arrow Datasets.
So my purpose for this issue is to see whether such usage of Arrow Datasets is desirable and / or possible, and if so, what requirements kartothek would have (potential missing features in Arrow Datasets).