python - 将 Python 列表拆分为具有最大内存大小的 block

Question

给定一条 python list的 bytes值(value)观:

# actual str values un-important
[
    b'foo',
    b'bar',
    b'baz',
    ...
]

如何将列表分成 block ，其中每个 block 的最大内存大小低于某个上限？

例如:如果上限是 7 个字节，那么原始列表将被分解为列表列表

[
    [b'foo', b'bar'], # sublist 0
    [b'baz'], # sublist 1
    ...
]

根据列表内容的累积长度，每个子列表最多为 7 个字节。

注意:每个子列表应该按照原始列表的顺序最大程度地打包。在上面的示例中，前 2 个 str 值被分组，因为它是 7 字节限制下可能的最大值。

预先感谢您的考虑和回复。

Answer 1

可以贪婪地解决序列的最优拆分问题，以使元素满足给定的最大/最小条件同时保持元素的顺序。
因此，您只需对输入序列进行一次迭代并维护一个元素缓冲区。
在 Python 中，这可以用生成器优雅地编码，这将具有不需要创建结果的优点。

您的问题的大部分算法如下:

def split_by_size(items, max_size, get_size=len):
    buffer = []
    buffer_size = 0
    for item in items:
        item_size = get_size(item)
        if buffer_size + item_size <= max_size:
            buffer.append(item)
            buffer_size += item_size
        else:
            yield buffer
            buffer = [item]
            buffer_size = item_size
    if buffer_size > 0:
        yield buffer

其中最后一个参数将确定给定项目大小的问题委托(delegate)给指定的可调用对象。
我不会详述这一点，但我会假设一个简单的 len()会做。
此外，这假设每个元素单独满足条件，否则也应该处理这种情况。

测试上面的代码:

import random


k = 10
n = 15
max_size = 10

random.seed(0)
items = [b'x' * random.randint(1, 2 * k // 3) for _ in range(n)]
print(items)
# [b'xxxx', b'xxxx', b'x', b'xxx', b'xxxxx', b'xxxx', b'xxxx', b'xxx', b'xxxx', b'xxx', b'xxxxx', b'xx', b'xxxxx', b'xx', b'xxx']

print(list(split_by_size(items, k)))
# [[b'xxxx', b'xxxx', b'x'], [b'xxx', b'xxxxx'], [b'xxxx', b'xxxx'], [b'xxx', b'xxxx', b'xxx'], [b'xxxxx', b'xx'], [b'xxxxx', b'xx', b'xxx']]

此外，如果您愿意将拆分结果存储在 list无论如何，上述方法的代码可以稍微紧凑一些:

def chunks_by_size(items, max_size, get_size=len):
    result = []
    size = max_size + 1
    for item in items:
        item_size = get_size(item)
        size += item_size
        if size > max_size:
            result.append([])
            size = item_size
        result[-1].append(item)
    return result

但也稍慢(见下面的基准)。

您也可以考虑使用 functools.reduce() (与 @NizamMohamed answer 基本相同)，代码会更短，但也可能更不可读:

def chunks_by_size_reduce(items, size, get_size=len):
    return functools.reduce(
        lambda a, b, size=size:
            a[-1].append(b) or a
            if a and sum(get_size(x) for x in a[-1]) + get_size(b) <= size
            else a.append([b]) or a, items, [])

当然效率不如 get_size()正在为所考虑的每个元素的“候选”内部列表的每个元素调用，这使得 O(n k!) , k是每个子序列中元素的平均数量。对于某些时间，请参阅下面的基准。

我不会对使用 itertools.accumulate() 的解决方案感到惊讶，但这也必然会很慢。

加快速度的最简单方法是使用 Cython或 Numba .
在这里，这适用于 split_by_size() .
对于他们俩来说，代码都不会改变。

对我们获得的所有这些进行基准测试( _cy 代表 Cython 编译的版本，而 _nb 代表 Numba 编译的版本):

%timeit list(split_by_size(items * 100000, k + 1))
# 10 loops, best of 3: 281 ms per loop
%timeit list(split_by_size_cy(items * 100000, k + 1))
# 10 loops, best of 3: 181 ms per loop
%timeit list(split_by_size_nb(items * 100000, k + 1))
# 100 loops, best of 3: 5.17 ms per loop
%timeit chunks_by_size(items * 100000, k + 1)
# 10 loops, best of 3: 318 ms per loop
%timeit chunks_by_size_reduce(items * 100000, k + 1)
# 1 loop, best of 3: 1.18 s per loop

请注意，虽然 Numba 编译的版本比其他版本快得多，但它也是最脆弱的，因为它需要 forceobj标志设置为 True ，这可能会导致执行不稳定。

无论如何，如果最终目标是通过一些 I/O 操作发送分组项目，我几乎不相信这将是一个瓶颈。

请注意，该算法与其他答案几乎相同，我只是发现这里的代码更简洁一些。