Does ddf.pipe() make sense?

[hussainsultan]

Often times, i end up writing a function that takes in a dask.dataframe. pandas implements pd.pipe(func) that i find pretty convenient.

This is a pretty easy to implement but i think pipe may be pretty confusing in dask.dataframe world especially if someone tries to do columnar reductions. Thoughts on if it makes sense to implement?

[mrocklin]

cc @TomAugspurger

[shoyer]

pipe is just sugar that lets you apply a function like a method. So, it seems pretty obvious and sensible to me.

This is a pretty easy to implement but i think pipe may be pretty confusing in dask.dataframe world especially if someone tries to do columnar reductions.

I'm not sure I understand the concern here. How would this be more confusing for dask.dataframe than for pandas?

[TomAugspurger]

I think this makes sense too.

pipe is just sugar that lets you apply a function like a method

Would dask expect the function to be a delayed function? Or would pipe accept any function that takes adask.dataframe?

[hussainsultan]

consider this scenario:

import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({'x': [1, 2, 3, 4, 5],
                          'y': [1., 2., 3., 4., 5.]})

ddf = dd.from_pandas(df, npartitions=2)
def f(df):
    return df.x.sum()

ddf.pipe(f).compute()

0    6
1    9
dtype: int64

while pandas pipe returns:

15

What should be the correct behavior here?

[mrocklin]

I think that dask.dataframe should almost always follow pandas semantics

[hussainsultan]

in that case, it makes sense for the input function to be a delayed method. Thoughts?

ddf.pipe(delayed(f)).compute()

[shoyer]

I would literally copy the implementation of pipe from pandas, e.g.,

    def pipe(self, func, *args, **kwargs):
        if isinstance(func, tuple):
            func, target = func
            if target in kwargs:
                raise ValueError('%s is both the pipe target and a keyword '
                                 'argument' % target)
            kwargs[target] = self
            return func(*args, **kwargs)
        else:
            return func(self, *args, **kwargs)

df.pipe(func) should not separately map func over partitions no more than calling a func(df) does.