如何将mmap.mmap.seek(pos)定位到文本文件任意第N行的起始位置?
场景1:仅单次查询指定行的起始位置
如果只需要偶尔查询某一行的位置,不需要反复查询不同行,直接遍历跳过目标行之前的所有行,用mmap自带的tell()方法获取当前字节偏移即可,遍历过程不会加载整个文件到内存,适配超大文件场景。
代码实现:
import mmap import os fbuff = open("big_file.csv", mode="r", encoding="utf8") f1_mmap = mmap.mmap(fbuff.fileno(), length=os.path.getsize("big_file.csv"), access=mmap.ACCESS_READ, offset=0) def get_target_line_pos(mmap_obj, target_line_no): # 此处target_line_no按1开始计数,第1行为表头行 mmap_obj.seek(0) # 跳过前 target_line_no -1 行 for _ in range(target_line_no - 1): line = mmap_obj.readline() if not line: # 行号超出文件总行数,返回异常值 return -1 # 当前指针位置即为目标行的起始字节偏移 return mmap_obj.tell() # 示例:获取第102457行的起始位置 line_pos = get_target_line_pos(f1_mmap, 102457) if line_pos != -1: f1_mmap.seek(line_pos) # 读取目标行,返回的bytes可按需decode为字符串 target_line_content = f1_mmap.readline().decode('utf8')
场景2:需要多次查询不同行的位置
如果需要频繁查询多个不同行的内容,建议一次性遍历文件构建行偏移索引,后续查询可以O(1)直接获取对应行的起始位置,避免多次遍历消耗IO。
代码实现:
def build_line_offset_index(mmap_obj): offset_list = [] mmap_obj.seek(0) while True: # 记录当前行的起始字节偏移 offset_list.append(mmap_obj.tell()) line = mmap_obj.readline() if not line: break return offset_list # 构建索引,仅需执行一次 line_index = build_line_offset_index(f1_mmap) # 示例1:查询第102457行(行号从1开始,对应索引下标为102456) pos = line_index[102456] f1_mmap.seek(pos) line_102457 = f1_mmap.readline().decode('utf8') # 示例2:查询第233行 pos = line_index[232] f1_mmap.seek(pos) line_233 = f1_mmap.readline().decode('utf8')
注意事项
- 行号计数规则:上述代码默认行号从1开始计数(第一行是表头),如果你按0开始计数,调整循环次数和索引下标即可。
- mmap的
readline()返回值为bytes类型,如果你需要字符串格式,调用decode('utf8')转换即可,编码和你打开文件时的参数保持一致。 - 两种方案都不会将全量文件加载到内存,完全适配内存无法容纳的超大文件场景。
内容的提问来源于stack exchange,提问作者Naveen Reddy Marthala
相关产品推荐
相关产品推荐

