从大文件中随机选择N行Python（无重复行）

Question

我需要使用Python从大型txt文件中提取N行数据。这些文件基本上是制表符分隔的表格。我的任务有以下限制：

• 这些文件可能包含标题（有些具有多行标题）。
• 标题需要按照相同的顺序出现在输出中。
• 每行只能被提取一次。
• 目前最大的文件约为150GB（约6000万行）。
• 在一个文件中，行大致相同长度，但在不同的文件中可能会有所不同。
• 通常我会提取5000个随机行（我可能需要高达100万行）

目前，我已经编写了以下代码：

inputSize=os.path.getsize(options.input)
usedPositions=[] #Start positions of the lines already in output

with open(options.input) as input:
    with open(options.output, 'w') as output:

        #Handling of header lines
        for i in range(int(options.header)):
            output.write(input.readline())
            usedPositions.append(input.tell())

        # Find and write all random lines, except last
        for j in range(int(args[0])):
            input.seek(random.randrange(inputSize)) # Seek to random position in file (probably middle of line)
            input.readline() # Read the line (probably incomplete). Next input.readline() results in a complete line.
            while input.tell() in usedPositions: # Take a new line if current one is taken
                input.seek(random.randrange(inputSize))
                input.readline() 
            usedPositions.append(input.tell()) # Add line start position to usedPositions
            randomLine=input.readline() # Complete line
            if len(randomLine) == 0: # Take first line if end of the file is reached
                input.seek(0)
                for i in range(int(options.header)): # Exclude headers
                    input.readline()
                randomLine=input.readline()
            output.write(randomLine)            

这段代码似乎正常工作。

我知道这段代码更喜欢处理紧随输入中最长行后的行，因为seek()最有可能返回最长行的位置，并且下一行会被写入输出。但是，由于输入文件中的行大致相同长度，这与此无关。 此外，我知道如果N大于输入文件中的行数，这段代码会导致无限循环。我不会实施此检查，因为获取行数需要很长时间。

RAM和HDD的限制并不重要。我只关心程序的速度。是否有方法进一步优化此代码？或者也许有更好的方法吗？

编辑：澄清一下，一个文件中的行大致相同长度。但是，我有多个文件需要运行此脚本，每个文件的平均行长度都不同。例如，文件A的每行大约有100个字符，而文件B的每行大约有50000个字符。我不知道任何文件的平均行长度。

- FableBlaze

---

这个问题更适合在http://codereview.stackexchange.com/上讨论。 - user647772

至少相关：Python从子文件夹中随机获取行 - Martijn Pieters

你需要从每个文件中获取x行随机行，还是从所有文件的所有行中获取？换句话说，你是从文件1中取10行，从文件2中取10行等等，还是从所有文件中随机取5000行？当你说“标题需要按相同顺序出现在输出中”时，你似乎跳过了它们，请问这是什么意思？ - Martijn Pieters

脚本需要仅使用单个输入文件进行操作，即从文件F中读取N行数据（这两个参数在调用文件时给出）。输入文件可能包含标题行（由一个参数指定其数量）。假设我有3个标题行，则在程序开始时，我只需将这三行写入输出。之后，我会跳过它们，因为我不希望它们出现在随机行的中间。 - FableBlaze

Answer 1

只有一种避免顺序读取文件直到抽样的最后一行的方法-我惊讶的是迄今为止没有一个答案提到过它：你需要在文件内随机寻找一个位置，读取一些字节，如果你有一个典型的行长度，就像你所说的，3或4倍的值应该就可以了。然后，在新的行字符（“\n”）上分割读取的块，并选择第二个字段-也就是随机位置的一行。

此外，为了能够一致地查找文件，它应该在“二进制读取”模式下打开，因此，必须手动处理换行标记的转换。

这种技术不能给你读取的行号，因此你需要保留文件中选定的行偏移量以避免重复。

#! /usr/bin/python
# coding: utf-8

import random, os


CHUNK_SIZE = 1000
PATH = "/var/log/cron"

def pick_next_random_line(file, offset):
    file.seek(offset)
    chunk = file.read(CHUNK_SIZE)
    lines = chunk.split(os.linesep)
    # Make some provision in case yiou had not read at least one full line here
    line_offset = offset + len(os.linesep) + chunk.find(os.linesep) 
    return line_offset, lines[1]

def get_n_random_lines(path, n=5):
    lenght = os.stat(path).st_size
    results = []
    result_offsets = set()
    with open(path) as input:
        for x in range(n):
            while True:
                offset, line = pick_next_random_line(input, random.randint(0, lenght - CHUNK_SIZE))
                if not offset in result_offsets:
                    result_offsets.add(offset)
                    results.append(line)
                    break
    return results

if __name__ == "__main__":
    print get_n_random_lines(PATH)

- jsbueno

---

我喜欢这个！我有一个类似的问题，我曾经相信我必须读两次文件。 - David M

我可能错了，但根据你的描述，我认为随机选择第一行是不可能的，因为你总是从随机选择的位置向前获取一个块，并使用下一行。 - David M

它可以平衡和微调，以便第一行和最后一行被选中的机会相同 - 这并不难。将文件导入索引的sqlite表中，并使用sql知道文件大小并检索随机行仍然更好。 - jsbueno

Answer 2

如果您需要从文件中获得N行的均匀样本，则需要知道要选择的确切行数；随机查找不会做到这一点，较长的行会使结果偏向于直接跟在最长行后面的行。

幸运的是，您只需要读取文件一次即可选择那些N行。您基本上选择您的N个第一行（以随机顺序），然后根据已读取的行数使用递减概率随机替换选定的行。

对于N == 1，第n行被替换为先前随机选择的概率为randint(0, n) < 1，因此，第二行有50％的机会被选中，第三行有33.33％的机会，等等。对于较大的N，在读取更多行时，使用相同的分布随机替换已选择集合中的一个已选择行。

在Python random lines from subfolders中，Blkknght编写了一个非常有用的函数，用于从可迭代对象中选择大小为N的随机样本。

import random

def random_sample(n, items):
    results = []

    for i, v in enumerate(items):
        r = random.randint(0, i)
        if r < n:
            if i < n:
                results.insert(r, v) # add first n items in random order
            else:
                results[r] = v # at a decreasing rate, replace random items

    if len(results) < n:
        raise ValueError("Sample larger than population.")

    return results

这很容易与您保存一组标题的要求相结合：

from itertools import islice

with open(options.input) as input:
    with open(options.output, 'w') as output:

        # Handling of header lines
        # Use islice to avoid buffer issues with .readline()
        for line in islice(input, int(options.header)):
            output.write(line)

        # Pick a random sample
        for line in random_sample(int(args[0]), input):
            output.write(line)

这个功能将会一次读取整个文件，选择一个均匀的随机样本，并将其写入输出文件。因此，该操作具有Θ(L)复杂度，其中L是文件中行数。

- Martijn Pieters

---

幸运的是，在我的当前任务背景下，我可以接受略微偏斜的结果。但如果我需要一个均匀的样本，这似乎是可行的方法。我给它点赞。 - FableBlaze

Answer 3

我认为更快的方法是随机选择N个行号，然后逐行查找文件并获取在列表中的行。目前你需要为每个随机数寻找随机位置，因此复杂度为 O(N*M)，其中M为文件大小。我建议使用O(M)的方法。

- zenpoy

---

我不知道行数，因此这种方法要求我两次循环输入文件。第一次获取行数，第二次打印选定的行。基本上是两次读取所有行，并在第二次决定该行是否输出（是否在随机数列表中）。我会测试它，因为编写不需要太长时间，但我怀疑两次读取所有行是否比N个（理想情况下）查找更快。 - FableBlaze

也许你可以通过将文件大小除以平均每行的大小（你说你知道）来估计行数，无论如何，两次遍历文件仍然是O(M)。 - zenpoy

我觉得我在行长方面表达得有些不太好。我知道一个文件中的行大致相同长度。然而，文件A和B之间的平均行长度可能会有所不同（例如100和50000个字符）。已相应地编辑了问题。 - FableBlaze

测试了从大约60000000行中选择5000个随机行的算法。我的原始算法要快得多。如果你仔细想想，这其实是有道理的。我的算法只需要大约5000次迭代。而循环遍历整个文件需要大约6000万次迭代（每行一个）。如果我计算一下行数，那么总共需要大约1.2亿次迭代。 - FableBlaze

@anti666：你只需要读取文件一次，看看我的答案。如果你可以接受加权结果（直接跟在更长的行后面的行更有可能被选中），那么就使用你自己的函数。 - Martijn Pieters

Answer 4

明显的改进是使用set()来替代你的usedPositions变量 - 查找速度会更快，而且由于你需要处理高达10^6个已用位置，查找时间也不可忽视。

在for循环中使用xrange而不是range。分配整个整数列表似乎并不必要。

- Abgan

---

谢谢！我会尝试包含这些优化。 - FableBlaze

Answer 5

未经测试（需要两次读取文件）：

import random

N = 5000
with open('file.in') as fin:
    line_count = sum(1 for i in fin)
    fin.seek(0)
    to_take = set(random.sample(xrange(line_count), N))
    for lineno, line in enumerate(fin):
        if lineno in to_take:
            pass # use it

然而，既然您提到行“大致”相同，那么您可以使用os.path.getsize并将其除以平均行长度（无论是已知的还是从文件中的N行中嗅探出来的），然后使用它来生成line_count - 对于一个随机样本来说，这应该足够接近。

您还可以mmap文件，并使用文件大小、平均行长度、最佳猜测的行数和随机行号来“查找”，然后向前或向后搜索下一行的开头。（由于mmap将使您能够像处理字符串一样处理它，因此您将能够使用带有偏移量的.index或者如果您真的想要的话，使用re）。

- Jon Clements

---

类似于zenpoy的建议。我将不得不研究mmap。 - FableBlaze

读取整个文件需要很长时间。单次循环超过60000000的时间比从同一文件中选择5000行时更长。 - FableBlaze