(self, func, *args, **kwargs)
return Streaming(stream, example, stream_type=stream_type)
Accumulate a function with state across batch elements

        See Also
        --------
        Streaming.map_partitions
Accumulate a function with state across batch elements
Accumulate a function with state across batch elements
def accumulate_partitions(self, func, *args, **kwargs):
        """ Accumulate a function with state across batch elements

        See Also
        --------
        Streaming.map_partitions
        """
        start = kwargs.pop('start', core.no_default)
        returns_state = kwargs.pop('returns_state', False)
        example = kwargs.pop('example', None)
        stream_type = kwargs.pop('stream_type', self._stream_type)
        if example is None:
            example = func(start, self.example, *args, **kwargs)
        if returns_state:
            _, example = example
        stream = self.stream.accumulate(func, *args, start=start,
                                        returns_state=returns_state, **kwargs)

        s_type = get_stream_type(example, stream_type)
        if s_type:
            return s_type(stream, example)
        return Streaming(stream, example, stream_type=stream_type)
def accumulate_partitions ( self , func , * args , * * kwargs ) : start = kwargs . pop ( 'start' , core . no_default ) returns_state = kwargs . pop ( 'returns_state' , False ) example = kwargs . pop ( 'example' , None ) stream_type = kwargs . pop ( 'stream_type' , self . _stream_type ) if example is None : example = func ( start , self . example , * args , * * kwargs ) if returns_state : _ , example = example stream = self . stream . accumulate ( func , * args , start = start , returns_state = returns_state , * * kwargs ) s_type = get_stream_type ( example , stream_type ) if s_type : return s_type ( stream , example ) return Streaming ( stream , example , stream_type = stream_type )