#NEWLINE# def apply( self , func , columns = no_default ) : #NEWLINE# #INDENT# ' Parallel version of pandas GroupBy.apply\n\n        This mimics the pandas version except for the following:\n\n        1.  The user should provide output columns.\n        2.  If the grouper does not align with the index then this causes a full\n            shuffle.  The order of rows within each group may not be preserved.\n\n        Parameters\n        ----------\n\n        func: function\n            Function to apply\n        columns: list, scalar or None\n            If list is given, the result is a DataFrame which columns is\n            specified list. Otherwise, the result is a Series which name is\n            given scalar or None (no name). If name keyword is not given, dask\n            tries to infer the result type using its beggining of data. This\n            inference may take some time and lead to unexpected result\n\n        Returns\n        -------\n        applied : Series or DataFrame depending on columns keyword\n        ' #NEWLINE# if ( columns is no_default ) : #NEWLINE# #INDENT# msg = "columns is not specified, inferred from partial data. Please provide columns if the result is unexpected.\n  Before: .apply(func)\n  After:  .apply(func, columns=['x', 'y']) for dataframe result\n  or:     .apply(func, columns='x')        for series result" #NEWLINE# warnings . warn ( msg ) #NEWLINE# dummy = self . _head ( ) . apply ( func ) #NEWLINE# columns = ( dummy . columns if isinstance ( dummy , pd . DataFrame ) else dummy . name ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# dummy = columns #NEWLINE# columns = self . _slice #NEWLINE# #UNINDENT# if isinstance ( self . index , Series ) : #NEWLINE# #INDENT# if ( self . index . _name == self . obj . index . _name ) : #NEWLINE# #INDENT# df = self . obj #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# df = self . obj . set_index ( self . index , drop = False , ** self . kwargs ) #NEWLINE# #UNINDENT# return map_partitions ( _groupby_apply_level0 , dummy , df , columns , func ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# from . shuffle import shuffle #NEWLINE# df = shuffle ( self . obj , self . index , ** self . kwargs ) #NEWLINE# return map_partitions ( _groupby_apply_index , dummy , df , self . index , columns , func )