在 Python 中使用随机化减少文本文件

Question

我在 bash 中解决了以下问题，但考虑到我需要减少的文件大小，我觉得它效率很低而且速度很慢。希望有人知道如何在 Python 中做同样的事情并希望加快速度。

最初的问题是减少非常大的文本文件（50-6000 万行，制表符分隔的列）。 其中一列被视为键，即我们确定文件中有多少具有唯一键的行，然后随机选择其中的一个百分比（例如，如果减少 75%，则为总数的四分之一）附加到一个新文件，将保留我们的结果。我们继续检查其余的键，随机化然后将包含每个唯一键的所有行减少相同的百分比。如果无法完成缩减 - 我们只需将所有行转移到生成的文件中。

正如我所说，我的 bash 脚本运行良好，但速度很慢，并且将各种 awk 和 grep 结构串在一起。无论如何，Python 应该以一种更优雅的方式来处理这个问题，并且不会过多地牺牲内存（同样，在这种情况下，我们正在处理 50+ 百万行文件）。 任何建议/技巧都会有所帮助！谢谢！

Answer 1

简单的解决方案是按键列对文件进行排序，例如，sort 第二列的制表符分隔输入：

#!/bin/bash
printf "a\tz\nb\ty\nc\tx" | sort -k 2 -t $'\t'

然后解决一个更简单的问题，即为每个唯一键检索 25% 的随机行，其中具有相同键的所有行都相邻，并且每个唯一键至少应保留一行：

#!/usr/bin/env python
import random
import sys
from itertools import chain, groupby

def choose_random(iterator, fraction, random=random.random):
    """Lazy analog of:

        L = list(iterator)
        k = int(len(L) * fraction + .5) or 1 # get at least one
        result = random.sample(L, k)

    Note: this function doesn't randomize the order of elements
          that would require to keep selected elements in memory
          and number of output elements is not exactly k
    """
    # always yield at least one item if input is not empty
    item = next(iterator)
    it = (x for x in chain([item], iterator) if random() < fraction)
    for x in chain([next(it, item)], it):
        yield x

def getkey(line):
    return line.split("\t")[1] # 2nd column

for key, group in groupby(sys.stdin, key=getkey):
    sys.stdout.writelines(choose_random(group, fraction=0.25))

注意：输入文件的最后一行应该包含换行符，否则如果选择了最后一行，则输出会损坏。

脚本接受标准输入上的排序（按键列）输入，并将减少的输出打印到标准输出。它一次只需要在内存中存储一​​行。这是一个单遍算法（O（n））。

Answer 2

因为你的问题比较模糊，所以我给出一个高层次的解决方案

1. 不要读取内存fileObj.read() 或fileObj.readlines() 中的整个文件，而是遍历文件for line in fileObj。为什么？这将是美好的回忆

2. 根据列表创建队列的实现

   class Queue(object):
       def __init__(self, max_size):
           self.queue = []
           self.max_size = max_size
       def __getitem__(self, index):
           if 0 <= index < max_size:
               return self.queue[index]
           else:
               raise IndexError
       def __iter__(self):
           return iter(self.queue)
       def push(seq):
           if isinstance(seq, Iterable):
               if len(self.queue) + len(seq) > self.max_size:
                   raise Full
               self.queue = seq
           else:
               if len(self.queue) + 1 > self.max_size:
                   raise Full
               self.queue.append(seq)
       def pop():
           if self.queue:
               return self.queue.pop(0)
   

3. 创建一个队列字典，其中 maxsize = 2 * 所选项目的百分比

类似

    PCT_SELECTED = 100
    MAXSIZE = 2 * PCT_SELECTED
    KEY_START = 10
    KEY_STOP = 15 
    from collection import defaultdict
    queue_dict = defaultdict(Queue(MAXSIZE))

1. 以非阻塞方式将元素放入队列中
2. 如果 Queue 已满，则会引发异常 Full，在这种情况下，您会从 Queue 中随机选择 50% 的元素并丢弃其余元素。

类似

    with open("your-file") as fin:
        for line in fin:
            key = line[KEY_START: KEY_STOP]
            try:
                queue_dict[key].push(line)
            except Full:
                queue_dict[key] = random.sample(queue_dict[key], PCT_SELECTED)

1. 最后遍历字典，随机修剪掉队列

   queue_dict = {key: random.sample(value, PCT_SELECTED) for key, value in queue_dict.items()}
   

2. 现在您可以通读字典并写入文件。

Answer 3

对于大量项目，只需检查每个项目的随机数即可选择 75%。

import random

with open('input') as f:
for line in f:
    if random.random() < 0.75:
        print line

如果您需要保证每个键至少有一项（即使它只有两行）：

import random
keys = set()

with open('input') as f:
    for line in f:
        columns = line.split('\t')
        key = columns[0]

        if not key in keys:
           print line
           keys.add(key)
           continue

        if random.random() < 0.75:
            print line