python:是否有用于分块输入流的库函数？

Question

我想将输入流分块以进行批处理.给定输入列表或生成器,

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

我想要一个返回该输入块的函数.说,如果chunk_size=4,那么,

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

这是我一遍又一遍地做的事情,并且想知道是否有比我自己写的更标准的方式.我错过了什么itertools吗？(人们可以用enumerate和解决问题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))

Answer 1

配方来自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)

Answer 2

[感谢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()包装来解决这个问题，但这不太漂亮。从概念上讲，使用repeatorcycle可能比count()但我出于某种原因进行字符计数更有意义。:^)

[FTR：不，我还没有完全认真地对待这个问题，但是嘿——今天是星期一。]