使用 itertools.islice 创建批次

Question

我正在尝试使用itertools.combinations 和itertools.slice 函数来创建可以并行执行计算的多个批处理。我使用以下函数来创建我的批次：

def construct_batches(n,k,batch_size):

    combinations_slices = []

    # Calculate number of batches
    n_batches = math.ceil(comb(n,k,exact=True)/batch_size)

    # Construct iterator for combinations
    combinations = itertools.combinations(range(n),k)

    while len(combinations_slices) < n_batches:
        combinations_slices.append(itertools.islice(combinations,batch_size))

    return combinations_slices

在执行了一些计算之后，我找出了哪些批次和元素是相关的。所以我有一个批次列表（例如batches = [2,3,1]）和一个元素列表（例如elements = [5,7,0]）。令我惊讶/恐怖的是，python 有以下行为。假设我想检查我的切片是否正确。那么

combinations_slices = construct_batches(n,k,batch_size)

list(combinations_slices[0])
Out[491]: 
[(0, 1, 2, 3),
 (0, 1, 2, 4),
 (0, 1, 2, 5),
 (0, 1, 2, 6),
 (0, 1, 2, 7),
 (0, 1, 2, 8),
 (0, 1, 2, 9),
 (0, 1, 3, 4),
 (0, 1, 3, 5),
 (0, 1, 3, 6)]

list(combinations_slices[1])
Out[492]: 
[(0, 1, 3, 7),
 (0, 1, 3, 8),
 (0, 1, 3, 9),
 (0, 1, 4, 5),
 (0, 1, 4, 6),
 (0, 1, 4, 7),
 (0, 1, 4, 8),
 (0, 1, 4, 9),
 (0, 1, 5, 6),
 (0, 1, 5, 7)]

这一切都很好，很愉快，表明该方法有效。但是，如果我使用列表理解将“相关”批次选择为combinations_slices = [combinations_slices[i] for i in range(len(combinations_slices)) if i in batches]，那么输出是（可悲的）：

combinations_slices = construct_batches(n,k,batch_size)

batches = [2,3,1]

combinations_slices = [combinations_slices[i] for i in range(len(combinations_slices)) if i in batches]

list(combinations_slices[0])
Out[509]: 
[(0, 1, 2, 3),
 (0, 1, 2, 4),
 (0, 1, 2, 5),
 (0, 1, 2, 6),
 (0, 1, 2, 7),
 (0, 1, 2, 8),
 (0, 1, 2, 9),
 (0, 1, 3, 4),
 (0, 1, 3, 5),
 (0, 1, 3, 6)]

list(combinations_slices[1])
Out[510]: 
[(0, 1, 3, 7),
 (0, 1, 3, 8),
 (0, 1, 3, 9),
 (0, 1, 4, 5),
 (0, 1, 4, 6),
 (0, 1, 4, 7),
 (0, 1, 4, 8),
 (0, 1, 4, 9),
 (0, 1, 5, 6),
 (0, 1, 5, 7)]

有什么方法可以在不将所有内容都转换为列表的情况下获得所需的行为（通常这些组合列表可能很大，所以我会用完内存......）？建议表示赞赏...

Answer 1

我对你的代码和描述有点困惑，但这里有一些指针。

more-itertools 库中有几个很好的批处理工具。看看分组部分中的chunk 和ichunk。您需要 pip install more-itertools 以使它们可用于您的代码。

ichunked 产生（惰性）islices 并返回一个生成器，因此它不应该使用太多内存。但是一旦你读取了一个 islice 或者消耗了生成器的输出，它们就会被耗尽并且不能再次迭代。

from more_itertools import ichunked

numbers = range(27)
for batch in ichunked(numbers, 5):
    print(batch)
print(ichunked(numbers, 5))

输出（分块）：

<itertools.islice object at 0x0000020BF63DD4F0>
<itertools.islice object at 0x0000020BF4519130>
<itertools.islice object at 0x0000020BF63DD4F0>
<itertools.islice object at 0x0000020BF4519130>
<itertools.islice object at 0x0000020BF63DD4F0>
<generator object ichunked at 0x00000261AFED7AC0>

chunked 如果您需要分析批次然后将它们提交到您的并行计算，可能会更有用。它将生成您可以多次阅读的列表，但由于它们是列表，它们将使用更多的内存：

from more_itertools import chunked

numbers = range(27)
for batch in chunked(numbers, 5):
    print(batch)
print(chunked(numbers, 5))

输出（分块）：

[0, 1, 2, 3, 4]
[5, 6, 7, 8, 9]
[10, 11, 12, 13, 14]
[15, 16, 17, 18, 19]
[20, 21, 22, 23, 24]
[25, 26]
<callable_iterator object at 0x000002163A25E370>

它还返回一个可调用的迭代器，它提供对元素的随机访问。这很有用，但会占用大量 RAM。您可以通过使用ichunked 并在相同数据上执行两次传递来绕过chunked 选项的大内存占用。比如：

def selected_batches()
    for batch_index, batchiter in enumerate(ichunked(numbers, 5)):
        if fulfils_criteria(batch):
            yield batch_index


def submit_to_parallel_computation():
    selected_batch_indexes = selected_batches()
    selected_batch_index = selected_batch_indexes.__next__()
    try:
        for batch_index, batchiter in enumerate(ichunked(numbers, 5)):
            if batch_index == selected_batch_index:
                add_to_parallel_work_queue(batchiter)
                # add_to_parallel_work_queue(list(batchiter))
                selected_batch_index = selected_batch_indexes.__next__()
    except StopIteration:
        # no more selected batches
        pass
    wait_for_results()

如果计算的分析/分块部分必须使用尽可能少的内存，并且与您将这些批处理提交到的并行计算相比，它在整个计算任务中所占的部分要小得多，那么这是一个很好的匹配。它还依赖于原始数据源本身，您可以多次迭代。

如果不是这种情况（因此您不能重复使用输入数据），您可以考虑创建一个中间文件。