FEAT: Add destructure support for analtyic and reduction UDF #2487

icexelloss · 2020-10-21T20:40:33Z

What is this change

This PR is a follow up of #2473. In this PR, I added the support for destructure for analytics and reduction UDFs:

Support for groupby aggregation UDF
Support for window analytic UDF (unbounded window)
Support for window aggregation UDF (moving window)

Example:

groupby aggregation UDF

    Define and use an UDF with multiple return columns:                                                                                                                                                     
                                                                                                                                                                                                            
    >>> @reduction(                                                                                                                                                                                         
    ...     input_type=[dt.double],                                                                                                                                                                         
    ...     output_type=dt.Struct(['mean', 'std'], [dt.double, dt.double])                                                                                                                                  
    ... )                                                                                                                                                                                                   
    ... def mean_and_std(v):                                                                                                                                                                                
    ...     return v.mean(), v.std()                                                                                                                                                                        
    >>>                                                                                                                                                                                                     
    >>> # create aggregation columns "mean" and "std"                                                                                                                                                       
    >>> table = table.groupby('key').aggregate(                                                                                                                                                             
    ...     mean_and_std(table['v']).destructure()                                                                                                                                                          
    ... )

window analytic UDF

    >>> @analytic(                                                                                                                                                                                          
    ...     input_type=[dt.double],                                                                                                                                                                         
    ...     output_type=dt.Struct(['demean', 'zscore'], [dt.double, dt.double])                                                                                                                             
    ... )                                                                                                                                                                                                   
    ... def demean_and_zscore(v):                                                                                                                                                                           
    ...     mean = v.mean()                                                                                                                                                                                 
    ...     std = v.std()                                                                                                                                                                                   
    ...     return v - mean, (v - mean) / std                                                                                                                                                               
    >>>                                                                                                                                                                                                     
    >>> win = ibis.window(preceding=None, following=None, group_by='key')                                                                                                                                   
    >>> # add two columns "demean" and "zscore"                                                                                                                                                             
    >>> table = table.mutate(                                                                                                                                                                               
    ...     demean_and_zscore(table['v']).over(win).destructure()                                                                                                                                           
    ... )

window aggregation UDF

    >>> @reduction(                                                                                                                                                                                         
    ...     input_type=[dt.double],                                                                                                                                                                         
    ...     output_type=dt.Struct(['mean', 'std'], [dt.double, dt.double])                                                                                                                                  
    ... )                                                                                                                                                                                                   
    ... def mean_and_std(v):                                                                                                                                                                                
    ...     return v.mean(), v.std()                                                                                                                                                                        
    >>>                                                                                                                                                                                                     
    >>> # create aggregation columns "mean" and "std"                                                                                                                                                       
    >>> table = table.groupby('key').aggregate(                                                                                                                                                             
    ...     mean_and_std(table['v']).destructure()                                                                                                                                                          
    ... )
    >>> win = ibis.window(preceding=ibis.interval(hours=1), following=None, group_by='key', order_by='time')
    >>> # add two columns "mean" and "std"
    >>> table = table.mutate(                                                                                                                                                                               
    ...     mean_and_std(table['v']).over(win).destructure()                                                                                                                                           
    ... )

Note

A lot of these changes don't work in PySpark backend because the support for struct in pandas_udf is limited (Only works in elementwise UDF from 3.0)

Tests

Added tests in test_vectorized_udf.py

icexelloss · 2020-10-21T21:55:55Z

ibis/backends/pandas/aggcontext.py

        max_lookback=None,
+        output_type=None,


Added this so we can check what's the expected output_type

icexelloss · 2020-10-21T21:56:56Z

ibis/backends/pandas/aggcontext.py

+        # Instead, we need to use "apply", which can return a non
+        # numeric type, e.g, tuple of two double.
+        if isinstance(self.output_type, dt.Struct):
+            return grouped_data.apply(function, *args, **kwargs)


This is because transform cannot take a func that returns non numeric data.

icexelloss · 2020-10-21T21:57:48Z

ibis/expr/types.py

@@ -863,14 +863,22 @@ class StructColumn(AnyColumn, StructValue):
    pass  # noqa: E701,E302


-class DestructColumn(AnyColumn):
-    """ Class that represents a destruct column.
+class DestructValue(AnyValue):


Mimic the class hierarchy of Struct

jreback · 2020-10-23T13:55:51Z

ibis/backends/pandas/execution/generic.py

-        for metric in op.metrics
-    ]
+
+        if isinstance(metric, ir.DestructValue):


hmm can you make a single function to do this (e.g. you can still call coerce_to_dataframe), but conceptually this should be something like

pieces = [ coerce_to_output_type(metric, execute(metric, scope=scope, timecontext=timecontext, **kwargs)) for metric in op.metrics]

Good point. I made a new function coerce_to_output for this.

jreback · 2020-10-23T13:56:18Z

ibis/backends/pandas/execution/selection.py

@@ -146,8 +147,7 @@ def compute_projection_column_expr(
            name=result_name,
        )
    elif isinstance(expr, ir.DestructColumn):
-        result.columns = expr.type().names
-        return result
+        return coerce_to_dataframe(result, expr.type().names)


i think this can also be much cleaner if you do the above

jreback · 2020-10-23T13:56:44Z

ibis/backends/pandas/execution/window.py


    if not group_by and not order_by:
-        aggcontext = agg_ctx.Summarize()
+        aggcontext = agg_ctx.Summarize(parent=parent, output_type=output_type)


is output_type now required (wouldn't object if it was)

It is not. It probably should be but currently this is created here: https://github.com/ibis-project/ibis/blob/master/ibis/backends/pandas/core.py#L207 and the expr here can be a TableExpr so it's not clear what the output type even mean in this case..

jreback · 2020-10-23T13:57:00Z

ibis/backends/pandas/udf.py

@@ -249,7 +250,21 @@ def execute_udaf_node_groupby(op, *args, aggcontext, **kwargs):
            def aggregator(first, *rest):
                # map(next, *rest) gets the inputs for the next group
                # TODO: might be inefficient to do this on every call
-                return func(first, *map(next, rest))
+                result = func(first, *map(next, rest))


This I didn't change because this is inner loop and a bit too messy to make coerce_to_output to work with this

ok can you create an issue to fix this up (as your comment indicates)

Sounds good! Added #2496

jreback · 2020-10-23T13:57:42Z

ibis/expr/operations.py

        for e in self.by + self.metrics:
-            names.append(e.get_name())
-            types.append(e.type())
+            if isinstance(e, ir.DestructValue):


can you add a comment here for future readers

icexelloss · 2020-10-26T20:23:42Z

@jreback Thanks! I addressed all comments. This is ready for another look.

jreback · 2020-10-26T20:55:20Z

ibis/backends/pandas/udf.py

@@ -249,7 +250,21 @@ def execute_udaf_node_groupby(op, *args, aggcontext, **kwargs):
            def aggregator(first, *rest):
                # map(next, *rest) gets the inputs for the next group
                # TODO: might be inefficient to do this on every call
-                return func(first, *map(next, rest))
+                result = func(first, *map(next, rest))


ok can you create an issue to fix this up (as your comment indicates)

jreback · 2020-10-26T20:56:38Z

ibis/util.py

@@ -73,6 +74,36 @@ def is_one_of(values: Sequence[T], t: Type[U]) -> Iterator[bool]:
 all_of = toolz.compose(all, is_one_of)


+def coerce_to_dataframe(data: Any, names: List[str]) -> pd.DataFrame:


since this is only used in pandas backend can you put it right about coerce_to_output, or even better integrate so you don't need this (e.g. just use coerce_to_output), but this could be a helper. is this possible?

Actually, this is also used by the pyspark backend to handle struct output from UDFs, so I left it in top level (ibis.util). Do you think we should still move it to pandas backend namespace?

jreback · 2020-10-26T20:57:03Z

ibis/util.py

+    pd.DataFrame
+    """
+    if isinstance(data, pd.DataFrame):
+        result = data


yeah this is not a great function because it sometimes returns the same object and sometimes a new one.

Out of curiosity, why is it not good to return same object sometimes and new one some other time?

… passed in DataFrames

jreback · 2020-10-26T22:44:46Z

needs a whatsnew note

jreback · 2020-10-26T22:50:24Z

thanks @icexelloss pls do a followon for the whatsnew note

Add destruct support for analtyic and reduction UDF

65e824d

icexelloss changed the title ~~Add destruct support for analtyic and reduction UDF~~ FEAT: Add destructure support for analtyic and reduction UDF Oct 21, 2020

icexelloss requested a review from jreback October 21, 2020 20:57

Fix tests

f363838

icexelloss commented Oct 21, 2020

View reviewed changes

datapythonista added the udf Issues related to user-defined functions label Oct 22, 2020

Fix custom window tests

4961ad4

icexelloss added pandas The pandas backend pyspark The Apache PySpark backend labels Oct 22, 2020

jreback suggested changes Oct 23, 2020

View reviewed changes

icexelloss added 2 commits October 26, 2020 16:02

Refactor to coerce_to_output

a822293

Add docs and comment

1006315

jreback suggested changes Oct 26, 2020

View reviewed changes

icexelloss mentioned this pull request Oct 26, 2020

refactor: remove coerce_to_dataframe from execute_udaf_node_groupby in pandas backend #2496

Closed

Clarify that coerce_to_dataframe returns the original frame object if…

c87855b

… passed in DataFrames

jreback approved these changes Oct 26, 2020

View reviewed changes

jreback added this to the Next Feature Release milestone Oct 26, 2020

jreback merged commit 400b079 into ibis-project:master Oct 26, 2020

icexelloss mentioned this pull request Nov 2, 2020

FEAT: Add support for destruct for aggregation with no groupby #2511

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FEAT: Add destructure support for analtyic and reduction UDF #2487

FEAT: Add destructure support for analtyic and reduction UDF #2487

icexelloss commented Oct 21, 2020 •

edited

Loading

icexelloss Oct 21, 2020

icexelloss Oct 21, 2020

icexelloss Oct 21, 2020

jreback Oct 23, 2020

icexelloss Oct 26, 2020

jreback Oct 23, 2020

icexelloss Oct 26, 2020

jreback Oct 23, 2020

icexelloss Oct 26, 2020 •

edited

Loading

jreback Oct 23, 2020

icexelloss Oct 26, 2020

jreback Oct 26, 2020

icexelloss Oct 26, 2020

jreback Oct 23, 2020

icexelloss Oct 26, 2020

icexelloss commented Oct 26, 2020

jreback Oct 26, 2020

jreback Oct 26, 2020

icexelloss Oct 26, 2020

jreback Oct 26, 2020

icexelloss Oct 26, 2020

jreback commented Oct 26, 2020

jreback commented Oct 26, 2020

		@@ -73,6 +74,36 @@ def is_one_of(values: Sequence[T], t: Type[U]) -> Iterator[bool]:
		all_of = toolz.compose(all, is_one_of)


		def coerce_to_dataframe(data: Any, names: List[str]) -> pd.DataFrame:

FEAT: Add destructure support for analtyic and reduction UDF #2487

FEAT: Add destructure support for analtyic and reduction UDF #2487

Conversation

icexelloss commented Oct 21, 2020 • edited Loading

What is this change

Note

Tests

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

icexelloss Oct 26, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

icexelloss commented Oct 26, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jreback commented Oct 26, 2020

jreback commented Oct 26, 2020

icexelloss commented Oct 21, 2020 •

edited

Loading

icexelloss Oct 26, 2020 •

edited

Loading