能否用mmap增量构建文件？Python本地DB重复键索引疑问

我是Python新手，正尝试使用mmap构建用于处理海量数据的本地数据库，当前采用的写入代码如下：

import os
import mmap
import cPickle as pickle
import operator
from functools32 import lru_cache

with open(path + '.data', 'wb') as f:
    index = []
    for key, value in iter_kvs:
        index.append((key, f.tell()))
        mmdict.serialise(value, f)

with open(path + '.index','wb') as f:
    mmdict.serialise(index, f)

由于设备性能限制，我无法一次性处理并存储完整数据集。在此想咨询：

1. 是否可以通过mmap增量构建文件？
2. 若某次运行中已将键‘cat’对应值10写入索引，下次处理另一数据分区时再次遇到键‘cat’对应值15，原有索引会被更新还是新增一条？

解答 #

1. 完全可以用mmap实现增量构建文件 #

当然没问题，不过得调整你现有代码的一次性写入逻辑，适配增量场景：

• 改用追加模式打开文件：把.data文件的打开模式从wb（覆盖写入）改成ab（追加写入），这样每次运行都会在文件末尾新增数据，不会清空原有内容。
• 加载已有索引：在处理新数据前，先检查.index文件是否存在，如果存在就读取并加载现有索引列表，而不是每次都新建空列表。
• 动态调整mmap映射：打开文件后将其映射到内存，当追加内容后，需要用mmap.resize()调整映射大小，确保能覆盖新增的文件内容。

给你一个简化的增量实现示例：

import os
import mmap
import cPickle as pickle
from functools32 import lru_cache

data_path = path + '.data'
index_path = path + '.index'

# 加载已有索引或初始化空索引
if os.path.exists(index_path):
    with open(index_path, 'rb') as f:
        index = pickle.load(f)
else:
    index = []

# 以追加模式打开数据文件并创建mmap映射
with open(data_path, 'ab') as f:
    f.seek(0, os.SEEK_END)  # 定位到文件末尾
    with mmap.mmap(f.fileno(), length=0, access=mmap.ACCESS_WRITE) as mm:
        for key, value in iter_kvs:
            current_pos = mm.tell()  # 记录当前写入位置
            index.append((key, current_pos))
            # 序列化数据到mmap内存区域
            serialised_data = pickle.dumps(value)
            mm.write(serialised_data)
        mm.flush()  # 确保内存数据同步到磁盘

# 保存更新后的索引
with open(index_path, 'wb') as f:
    pickle.dump(index, f)

2. 当前逻辑会新增条目，不会自动更新 #

按照你现在的代码，每次运行都会新建空的index列表，处理完当前数据后直接覆盖写入.index文件。如果下次再遇到键cat，新的(cat, 新偏移量)会被加入新列表，最终覆盖原有索引，结果就是索引里会存在两条cat的条目（对应不同的数据位置）。

如果想要实现更新原有键的值，需要在处理前先加载已有索引并做键检查：

• 把索引转成字典（{key: 偏移量}），方便快速查找键是否存在；
• 遇到已存在的键时，直接更新字典里的偏移量，而非新增条目；
• 最后再把字典转回列表保存为新索引。

示例代码片段：

# 加载索引后转成字典
index_dict = {key: pos for key, pos in index}

for key, value in iter_kvs:
    current_pos = mm.tell()
    # 存在则更新偏移量，不存在则新增
    index_dict[key] = current_pos
    serialised_data = pickle.dumps(value)
    mm.write(serialised_data)

# 转回列表保存
index = list(index_dict.items())

这样处理后，每个键只会保留最新的偏移量，查询时就能拿到对应最新的值了。

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