如何在Python中按任意比例拆分流式数据？

你提到的通过计数器控制比例的方案属于误差补偿式分配，是这类需求下兼顾全局精度和局部均匀性的最优方案之一，比纯随机分配更适合小数据量场景，不会出现小样本下比例严重偏移的问题。

核心实现逻辑 #

只需要维护一个全局累计误差变量即可，不需要两个计数器，逻辑非常简单：

1. 设定目标比例 p（如15%即填0.15），初始化累计误差为0
2. 每处理一条数据，将 p 加到累计误差上
3. 如果累计误差≥1，就将当前数据分配到第一个流，同时累计误差减1；否则分配到第二个流

这个方案的特性完全匹配你的需求：

• 全局误差永远小于1，总数据量为N时，第一个流的数据量只会是floor(N*p)或ceil(N*p)，精度完全符合要求
• 局部分布均匀，不会出现连续多条数据都分配到同一个流的情况，处理固定大小的chunk时，每个chunk内的分配比例也会和目标比例非常接近，比如你举的15%比例、5条大小的chunk场景，每个chunk基本会分配1条到第一个流，偶尔出现0条的情况，总累计完全符合要求。

结合itertools的通用实现 #

你提到的partition完全可以作为实现基础，只需要把上面的误差判断逻辑做成partition需要的谓词函数即可，不需要额外写复杂的自定义逻辑。如果需要处理分块数据流，只需要把误差变量放到全局维护，跨chunk保持状态即可。

参考实现代码：

import itertools

def partition_by_ratio(ratio, iterable):
    """按指定比例将迭代器拆分为两个迭代器，ratio为分到第一个迭代器的比例"""
    error = 0.0
    def predicate(item):
        nonlocal error
        error += ratio
        if error >= 1:
            error -= 1
            return True
        return False
    # 此处partition逻辑对应itertools recipes中的标准实现
    t1, t2 = itertools.tee(iterable)
    return filter(predicate, t1), itertools.filterfalse(predicate, t2)

如果是处理已经分好块的buffered数据流，直接循环处理每个chunk，复用同一个谓词函数即可自动保持全局比例精度。

内容的提问来源于stack exchange，提问作者davidvandebunte