6

我想分块输入流以进行批处理。给定一个输入列表或生成器，

x_in = [1, 2, 3, 4, 5, 6 ...]

我想要一个可以返回该输入块的函数。说，如果chunk_size=4，那么，

x_chunked = [[1, 2, 3, 4], [5, 6, ...], ...]

这是我一遍又一遍地做的事情，并且想知道是否有比自己编写更标准的方法。我错过了什么itertools吗？enumerate（可以用and解决这个问题groupby，但感觉很笨拙。）如果有人想看到一个实现，这里是，

def chunk_input_stream(input_stream, chunk_size):
    """partition a generator in a streaming fashion"""
    assert chunk_size >= 1
    accumulator = []
    for x in input_stream:
        accumulator.append(x)
        if len(accumulator) == chunk_size:
            yield accumulator
            accumulator = []
    if accumulator:
        yield accumulator

编辑

受 kreativitea 回答的启发，这里有一个解决方案islice，它很简单且不需要后过滤，

from itertools import islice

def chunk_input_stream(input_stream, chunk_size):
    while True:
        chunk = list(islice(input_stream, chunk_size))
        if chunk:
            yield chunk
        else:
            return

# test it with list(chunk_input_stream(iter([1, 2, 3, 4]), 3))
4

3 回答 3

6

配方来自itertools：

def grouper(n, iterable, fillvalue=None):
    "Collect data into fixed-length chunks or blocks"
    # grouper(3, 'ABCDEFG', 'x') --> ABC DEF Gxx
    args = [iter(iterable)] * n
    return izip_longest(fillvalue=fillvalue, *args)
于 2012-11-05T20:13:40.687 回答
4

[感谢 OP 的更新版本：yield from自从我升级以来，我一直在看一切，我什至没有想到我在这里不需要它。]

哦，什么鬼：

from itertools import takewhile, islice, count

def chunk(stream, size):
    return takewhile(bool, (list(islice(stream, size)) for _ in count()))

这使：

>>> list(chunk((i for i in range(3)), 3))
[[0, 1, 2]]
>>> list(chunk((i for i in range(6)), 3))
[[0, 1, 2], [3, 4, 5]]
>>> list(chunk((i for i in range(8)), 3))
[[0, 1, 2], [3, 4, 5], [6, 7]]

chunk_input_stream警告：如果输入是列表，则上述问题与 OP 相同。你可以用一个额外的iter()包裹来解决这个问题，但这不那么漂亮。从概念上讲，使用repeator可能比但我出于某种原因计算字符cycle更有意义。count():^)

[FTR：不，我对这件事还不是很认真，但是嘿——今天是星期一。]

于 2012-11-05T21:12:24.683 回答
1

你有什么理由不使用这样的东西吗？：

# data is your stream, n is your chunk length
[data[i:i+n] for i in xrange(0,len(data),n)]

编辑：

由于人们正在制造发电机......

def grouper(data, n):
    results = [data[i:i+n] for i in xrange(0,len(data),n)]
    for result in results:
        yield result

编辑 2：

我在想，如果您将内存中的输入流作为双端队列，您可以.popleft非常有效地产生 n 个对象。 

from collections import deque
stream = deque(data)

def chunk(stream, n):
    """ Returns the next chunk from a data stream. """
    return [stream.popleft() for i in xrange(n)]

def chunks(stream, n, reps):
    """ If you want to yield more than one chunk. """
    for item in [chunk(stream, n) for i in xrange(reps)]:
        yield item
于 2012-11-05T20:26:06.170 回答