(func, *args, **kwargs)
return Streaming(stream, example, stream_type=stream_type)
Map a function across all batch elements of this stream

    The output stream type will be determined by the action of that
    function on the example

    See Also
    --------
    Streaming.accumulate_partitions
Map a function across all batch elements of this stream
Map a function across all batch elements of this stream
def map_partitions(func, *args, **kwargs):
    """ Map a function across all batch elements of this stream

    The output stream type will be determined by the action of that
    function on the example

    See Also
    --------
    Streaming.accumulate_partitions
    """
    example = kwargs.pop('example', None)
    if example is None:
        example = func(*[getattr(arg, 'example', arg) for arg in args], **kwargs)

    streams = [arg for arg in args if isinstance(arg, Streaming)]
    if 'stream_type' in kwargs:
        stream_type = kwargs['stream_type']
    else:
        stream_type = ('streaming'
                       if any(s._stream_type == 'streaming' for s in streams)
                       else 'updating')

    if len(streams) > 1:
        stream = type(streams[0].stream).zip(*[getattr(arg, 'stream', arg) for arg in args])
        stream = stream.map(apply_args, func, kwargs)

    else:
        s = streams[0]

        if isinstance(args[0], Streaming):
            stream = s.stream.map(func, *args[1:], **kwargs)
        else:
            other = [(i, arg) for i, arg in enumerate(args)
                     if not isinstance(arg, Streaming)]
            stream = s.stream.map(partial_by_order, function=func, other=other,
                                  **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 map_partitions ( func , * args , * * kwargs ) : example = kwargs . pop ( 'example' , None ) if example is None : example = func ( * [ getattr ( arg , 'example' , arg ) for arg in args ] , * * kwargs ) streams = [ arg for arg in args if isinstance ( arg , Streaming ) ] if 'stream_type' in kwargs : stream_type = kwargs [ 'stream_type' ] else : stream_type = ( 'streaming' if any ( s . _stream_type == 'streaming' for s in streams ) else 'updating' ) if len ( streams ) > 1 : stream = type ( streams [ 0 ] . stream ) . zip ( * [ getattr ( arg , 'stream' , arg ) for arg in args ] ) stream = stream . map ( apply_args , func , kwargs ) else : s = streams [ 0 ] if isinstance ( args [ 0 ] , Streaming ) : stream = s . stream . map ( func , * args [ 1 : ] , * * kwargs ) else : other = [ ( i , arg ) for i , arg in enumerate ( args ) if not isinstance ( arg , Streaming ) ] stream = s . stream . map ( partial_by_order , function = func , other = other , * * 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 )