6

バッチ処理のために入力ストリームをチャンクしたい。入力リストまたはジェネレーターが与えられると、

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))
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()が、それはあまりきれいではありません。概念的には、repeatまたはを使用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：

メモリー内に両端キューとして入力ストリームがある場合、.popleftn個のオブジェクトを非常に効率的に生成できると考えていました。 

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 に答える