Let's talk DataFrame APIs

[xdssio]

It is evident at this point that pandas had led the way in how people interact and expect DataFrames to behave and can be the baseline and the starting point for an API of the new age of more scalable feature riched and machine-learning-oriented DataFrames.
As PySpark tried to make a DataFrame for distributed work, we can learn from their work and adoption to see another approach. I will reference koalas as an attempt to back-track some of those decisions, as currently, it does not integrate as expected with PySpark itself and the conversions between them are cumbersome.
The last relevant references I'll mention is polars, which is the backbone engine for Daft, but shy away from distributed execution and currently needs to address machine-learning and, specifically, machine-learning pipelines needs with extensions and Ray, which is the distributed backbone for Daft.

Whenever deciding on an API, we need to remember the implications of the distributed implementation cost in development time and the attempt to keep the Daft APIs concise but expressive.

Another essential element to consider is the use of lazy evaluation in implementations and the use of UDFs, which can fill many transformations-needs so we can avoid implementing it.

Reference

• Daft user guide
• polars user guide
• pandas user guide

Let's review some standard APIs and open the discussion to new and missed ones.

Exploration

DataFrame

• head
  • limit
• tail
• __len__
• collect - What do you think about the good use cases to materialise the dataframe, and should we keep it in memory?
• sort
• __getitem__ - Do we prefer daft.col('a') vs df['a'] or df.a, which can be messed up with problematic column names but gives you auto-completion?
  Do we want to have this distinction between a series and an expression like Polars, or always use lazy expressions like Vaex?
• __setitem__ - is the use of with_column/ with_columns sufficient? or do we want to have df['a'] = df['b'] + 1 ?
• select columns
  • df.select(columns)
  • df.exclude(columns)
  • df[[col1,col2]]
  • df[int, str]
• select_rows - which API is better?
  • df.iloc[]
  • df[10:20]
• unique
• numunique
• is_na, is_null
• is_nan
• countna
• value_counts

Column (pandas collection)

• count
• min
• max
• mean
• cast
• expression operations (+,- etc.)
• datetime operations
• timedelta operations
• string operations
• image operations - should we only use UDFs here?

IO

• from_list (assuming key-value records)
• from_pydict (key- array)
  • should it be from_dict?
• read_csv
• read_parquet
• read_lance lance
• read_images (read_files)
• read_text (read_files?)
• read_json
• to_pandas
• to_ray
• to_arrow
• to_numpy
• from_pandas
• from_ray
• from_polars
• from_arrow
  • from_sparse
• SQL connections
  • BigQuery
  • Postgress
  • etc
• write_parquet
• write_csv

Data Manipulation

• apply
• map
• with_column
• with_columns
  -[ ] explode
• transpose / pivot
• NumPy functions
• fill_na
• filter
  • where
  • drop_duplicates (we have distinct)
  • drop_na

Time-series

• rolling

GroupBy aggregations

• mean
• sum
• count
• std
• distinct
• quantiles
• Custom functions

Combining

• join
• append / concat

ML

• split_random
  • repartition
• kfold

[xdssio]

Here is a great reference to the Consortium for Python Data API which also discuss some of this issues.

[jaychia]

Thanks for this @xdssio!

A few initial comments:

polars, which is the backbone engine for Daft

We actually run directly on Arrow instead of relying on Polars. Our current codebase does use certain functionality from PyArrow/Polars/Pandas, but we are moving into a Rust implementation which will allow us to write our own kernels and leverage Arrow2 where possible. At some point, PyArrow/Polars/Pandas will all be optional dependencies.

we need to remember the implications of the distributed implementation ... use of lazy evaluation in implementations and the use of UDFs, which can fill many transformations-needs so we can avoid implementing it

You're absolutely correct! Some things are trivial in a non-distributed world, but when dealing with a distributed dataframe may involve expensive shuffles and be prohibitively expensive. Additionally, being lazy can also lead to some unexpected behavior - for example the simple len() function is ubiquituous Python for getting the length of some in-memory datastructure, but calling len() and having it be blocking for minutes as our lazy dataframe computes could be unexpected behavior for users!

API Discussions

Here are my comments for DataFrame APIs. I'll also start converting some of these to issues/separate discussions that we can chip away at!

head and tail

Currently, .limit has the same behavior as a Pandas/Polars .head. We have no equivalent of a .tail, which actually might require some work on our scheduler to make work since we have a feed-forward scheduler that processes partitions from start to end.

Interestingly, in the distributed world, Spark actually has a different behavior than Pandas/Polars for these methods. Instead of returning a new DataFrame, it blocks on computation and then returns a list of Row objects to the client. Internally within the team, we have discussed having a row iterator interface that can accomplish something similar.

Additionally, PostgreSQL LIMITs do not support "tailing from the back". Instead, it allows for an offset when calling LIMIT.

Proposed Features:

1. Let's put .tail on the backburner for now - SQL itself doesn't seem to provide any builtin functionality for this. Here is the GitHub discussion feature thread to track this [Feature] Tailing DataFrames #474
2. I will open a GitHub Discussion for a row iterator interface so that users can stream rows of the DataFrame, similar to the Spark .head/.tail APIs. This is a feature that can be useful for things such as feeding data into ML training. [Feature] Row streaming from a DataFrame #475

len

Using Polars and Spark as references, they do not support calling len over a lazy dataframe. This makes sense as the length of a lazy (uncomputed) dataframe is by definition, undefined.

Understandably, this may cause some confusion as users are used to calling len() over a Pandas/eager-Polars DataFrame. In order to prevent unexpected behavior, I'd like to propose that len() never cause any side-effects of df execution.

Proposed Features:

#476

1. We implement a .count_rows() method that returns an integer count of the number of rows in the final dataframe.
2. Calling __len__ over a materialized DataFrame can return the length of the DataFrame. However, calling __len__ over a non-materialized DataFrame should raise an appropriate error informing users to either materialize the df first, or use .count_rows().

collect

.collect() is intended to be a way for users to interact with Daft in an interactive setting. By calling .collect(), the user is saying "materialize my current results so that everything up to this step is cached, and subsequent operations in my interactive session will be faster".

NOTE: Unlike a PySpark .collect(), .collect() does not pull data down to the caller in distributed execution. I.e. data is materialized in distributed memory (potentially out-of-core if it does not fit in distributed memory).

In an eager dataframe such as Pandas/eager-Polars, every operation on the DataFrame is collected, since every step is eagerly executed and materialized. Daft's .collect() mechanism is simply a way for users to control this materialization for a better interactive development experience.

# Eager DataFrame: blocks and materialized on every line
# Lazy DataFrame: only blocks on materializes on `.collect()` so that `print(df.count_rows())` and `df.write_csv(...)` can avoid recomputation

df = df.with_column(...)
df = df.where(...)
# for a lazy dataframe, call df.collect() here
print("Writing number of rows:", df.count_rows())
df.write_csv(...)

Perhaps we should rename .collect() to .execute() or something similar, to avoid confusion with Spark. I'll leave this as an open question for now.

Proposed Features:

1. We should have a better user guide on .collect() and how to use it in an interactive session.

getitem

1. daft.col("A") is useful for cases where you don't yet have a dataframe to reference a column, but want to write an expression. This is a bit of a rarer case but does come up from time to time.
2. df["A"] is our preferred way of referring to columns in a DataFrame since it is unambiguous during resolution.
3. df.a is ambiguous as it could refer to a column name or a method in a DataFrame. It does give some autocompletion benefits, but I think the downsides outweigh the benefits here. We like having one way of doing things (Zen of Python: There should be one-- and preferably only one --obvious way to do it.).

We actually don't yet have a Series abstraction, so there is no way to access a Series.

setitem

We have discussed this before internally within the team - this is actually surprisingly difficult to implement and get right. Happy to talk through it more if this is a hotly requested feature.

(edited)

select columns

df[[col1,col2]] should work as well actually. It more-or-less aliases df.select(col1, col2).

Daft does not have indices like Pandas, so df[int, str] is not a valid operation.

select rows

We like that dataframe indexing (df[...]) is explicit and unambiguous in referring to column selection at this moment.

Selection of rows can be achieved with .limit, and we are open to adding an offset option there to enable slicing. See discussion on head/tail.

unique, num_unique, value_counts, countna

These make sense, and should make it into the API. We currently only have a df.distinct() which behaves similarly to a SQL SELECT DISTINCT *.

Proposed Features:

#477

1. df.distinct() should take in column arguments to perform a DISTINCT operation on only a selected group of columns. This should cover the unique use-case as well where a user wants to retrieve the unique values for a given column/tuple of columns. []
2. num_unique is an aggregation - basically a customized count function. We should implement this similar to other aggregations such as mean/count
3. value_counts is an .agg([(c, "count_unique") for c in cols]) to return a DataFrame of unique columns + counts on their occurrences
4. countna is an aggregation as well, and can be implemented as such. Aliasing a sum of .is_null().cast(int) might do the trick

[jaychia]

Column Operations

We currently have a very limited set of operations, but would like to start adding a more comprehensive set of operations for:

1. Strings [Feature] String Operations #478
2. Dates and Datetimes [Feature] Datetime Operations #479

Image operations are coming soon - we have plans to write Arrow extension types for representing images, and can start to define kernels on these types. For now, UDFs are the main way of interacting with images.

I/O

We have a basic set of I/O operations, but are open to supporting more on a use-case driven basis.

Data manipulations

1. .map feature proposal: [Feature] Expression.map #480
2. We'd like to avoid with_columns for now to avoid feature bloat, but happy to explore if this can significantly reduce code verbosity
3. transpose - happy to explore use-cases for transpose, but we were not able to think of much uses for it.
4. pivot - this is a pretty advanced aggregation, we'll start a feature thread for it please comment on it if it needs to be prioritized: [Feature] Pivot aggregations #481
5. NumPy functions - interesting, not sure what this means but happy to learn more. Is this a .apply?
6. fillna - this could be useful, and would just alias df["col"].is_null().if_else(val, df["col"]). We need to be careful here about null semantics: [Feature] Add Expression helper to fill NA #482
7. dropna - this can just alias DataFrame.where(~df["col"].is_null()): [Feature] Drop NA rows #483

[dendihandian]

what is the equivalent syntax for this pandas syntax in daft: df = df.apply(lambda row: processing_row(row, sqlite_conn), axis=1)

[jaychia]

Thanks for the question! Answered in: #1041

[jaychia]

Rolling/Window

I will add a Discussion post on rolling/window operations, but we don't currently have any fleshed out specs around this. Would love to start discussing about the use-cases and thinking about potential implementations though: #484

Aggregations

I added a discussion thread on aggregations: #485

We can also discuss custom aggregations there. We'd like to understand the use-cases before designing an API around that.

Combining

Concat: #486

ML

Repartition is actually different from a split. We will probably want a dedicated split function for train/val sets that produce two separate dataframes based on some ratio of a split: #487