Skip to main content
 首页 » 编程设计

python之我可以使用 bisect 来打印一行的内容吗

2025年05月04日7luoye11

我有一个文件,其中每一行都按字母顺序排列。该文件是 12Gb,这意味着我不能简单地逐行读取它。数据如下所示:

brown    0    1    0    1    2 
fox      3    5    0    0    1 
jumped   2    0    6    1    0 

每行开头的单词都是唯一的。每行中的单词和数字由制表符分隔。我希望能够查询特定关键字的文件。例如,如果我查询“fox”,程序应该返回“fox 3 5 0 0 1”。

似乎对分模块是一个很好的候选者:https://docs.python.org/3.0/library/bisect.html

我找到了一个使用二分法找出关键字行号的帖子:How do I perform binary search on a text file to search a keyword in python?

代码如下:

import bisect 
import os 
 
class Query(object): 
    def __init__(self, query, index=5): 
        self.query = query 
        self.index = index 
 
    def __lt__(self, comparable): 
        return self.query < comparable[self.index:] 
 
class FileSearcher(object): 
    def __init__(self, file_pointer, record_size=35): 
        self.file_pointer = file_pointer 
        self.file_pointer.seek(0, os.SEEK_END) 
        self.record_size = record_size + len(os.linesep) 
        self.num_bytes = self.file_pointer.tell() 
        self.file_size = (self.num_bytes // self.record_size) 
 
    def __len__(self): 
        return self.file_size 
 
    def __getitem__(self, item): 
        self.file_pointer.seek(item * self.record_size) 
        return self.file_pointer.read(self.record_size) 
 
with open('myfile') as file_to_search: 
    query = 'fox\t' #token to query 
    wrapped_query = Query(query) 
    searchable_file = FileSearcher(file_to_search) 
    linepos = bisect.bisect(searchable_file, wrapped_query) 
    print "Located @ line: ", linepos 
    #print content of line? 

但是,我不知道如何实际打印该行的内容。我至少应该在某处添加一个读取语句,但我不知道在哪里。

是否可以用bisect模块打印行的内容?

请您参考如下方法:

如果你想使用 Python 解决方案,你可以执行以下操作:

  • 以 MAX_LINE 字节的小块读取文件,每次向前移动固定偏移量
  • 该偏移量决定了 block 的大小
  • 对于每个这样的阅读,确定关键(一行中的第一个单词)
  • 这些键作为 block 的分隔符
  • 构建此类键的列表。该列表将按键排序
  • 您可以通过 pickle/json.dumps/... 将此类列表保存在某处
  • 查询时,通过平分找到你key所在 block 的索引
  • 完整读取该 block 并找到包含数据的 key

这是示例文件 bigfile:

abc 4 
bar 2 
baz 3 
egg 6 
foo 1 
god 8 
ham 5 
sex 7 

代码:

import os 
from bisect import bisect 
 
MAX_LINE = 7 
BLOCK_SIZE = 10 
 
def parse_chunks(filename): 
 
    size = os.path.getsize(filename) 
    chunks = [] 
 
    with open(filename, 'rb') as file: 
        block = str(file.read(MAX_LINE*2)) 
        first_line = block[:block.find('\n') + 1] 
        chunks.append(first_line.split()[0]) 
 
        pos = BLOCK_SIZE 
        while pos < size: 
            file.seek(pos) 
            block = str(file.read(MAX_LINE*2)) 
            first_eol = block.find('\n') 
            second_eol = block.find('\n', first_eol + 1) 
            if first_eol == -1 or second_eol == -1: 
                break 
 
            line = block[first_eol + 1:second_eol] 
 
            key = line.split()[0] 
            chunks.append(key) 
 
            pos += BLOCK_SIZE 
 
    return chunks 
 
 
if __name__ == '__main__': 
    BLOCK_SIZE = 10 
    filename = 'bigfile' 
    chunks = parse_chunks(filename) 
 
    query = 'abc' 
    pos_before = bisect(chunks, query) - 1 
 
    with open(filename, 'rb') as file: 
        file.seek(pos_before*BLOCK_SIZE) 
        block = str(file.read(BLOCK_SIZE + MAX_LINE)) 
        line_start = block.find(query) 
        line_end = block.find('\n', line_start + 1) 
        line = block[line_start:line_end] 
 
        print(line) 

在这个玩具示例中,我使用 10 字节的 block 大小,对于 12GB 的文件,我建议您从 1M 开始。