Python 多处理 - 共享大型数据集

Question

我正在尝试加速受 CPU 限制的 Python 脚本（在 Windows11 上）。 Python 中的威胁似乎不在不同的 cpu（核心）上运行，所以我唯一的选择是多处理。

我有一个很大的字典数据结构（从文件加载后占用 11GB 内存），我正在检查计算值是否在该字典中。计算的输入也来自一个文件（大小为 100GB）。这个输入我可以分批池映射到进程，没问题。但是我不能将字典复制到所有进程，因为没有足够的内存。所以我需要找到一种方法让进程检查值（实际上是一个字符串）是否在字典中。

有什么建议吗？

伪程序流程：

--main--
- load dictionary structure from file   # 11GB memory footprint
- ...
- While not all chuncks loaded
-    Load chunk of calcdata from file   # (10.000 lines per chunk)
-    Distribute (map) calcdata-chunck to processes
-    Wait for processes to complete all chunks

--process--
- for each element in subchunk
-    perform calculation
-    check if calculation in dictionary  # here is my problem!
-    store result in file

编辑，在下面实现 cmets 之后，我现在在：

def ReadDictFromFile()
    cnt=0
    print("Reading dictionary from " + dictfilename)
    with open(dictfilename, encoding=("utf-8"), errors=("replace")) as f:
        next(f) #skip first line (header)
        for line in f:
            s = line.rstrip("\n")
            (key,keyvalue) = s.split()
            shared_dict[str(key)]=keyvalue
            cnt = cnt + 1
            if ((cnt % 1000000) == 0): #log each 1000000 where we are
                print(cnt)
                return #temp to speed up testing, not load whole dictionary atm
    print("Done loading dictionary")        


def checkqlist(qlist)
    try:
        checkvalue = calculations(qlist)
        
        (found, keyval) = InMem(checkvalue)
                
        if (found):
            print("FOUND!!! " + checkvalue + ' ' + keyvalue)            
    except Exception as e:
        print("(" + str(os.getpid()) + ")Error log: %s" % repr(e))
        time.sleep(15)


def InMem(checkvalue):
    if(checkvalue in shared_dict):
        return True, shared_dict[checkvalue]
    else:
        return False, ""


if __name__ == "__main__":
    start_time = time.time()

    global shared_dict 
    manager = Manager()
    shared_dict = manager.dict()

    ReadDictFromFile()
    with open(filetocheck, encoding=("utf-8"), errors=("replace")) as f:
        qlist = []
        for line in f:
            s = line.rstrip("\n")
            qlist.append(s)
            if (len(qlist) > 10000):
                try:
                    with multiprocessing.Pool() as pool:
                        pool.map(checkqlist, qlist)            
                except Exception as e:
                    print("error log: %s" % repr(e))
                    time.sleep(15)
    logit("Completed! " + datetime.datetime.now().strftime("%I:%M%p on %B %d, %Y"))
    print("--- %s seconds ---" % (time.time() - start_time))

Answer 1

你可以使用multiprocessing.Manager.dict，这是最快的IPC，你可以用来在python中的进程之间进行检查，对于内存大小，只需通过将所有值更改为None来使其更小，在我的电脑上它可以做到33k成员每秒检查一次……比普通字典慢约 400 倍。

manager = Manager()
shared_dict = manager.dict()
shared_dict.update({x:None for x in main_dictionary})
shared_dict["new_element"] = None  # to set another value
del shared_dict["new_element"]  # to delete a certain value

您还可以为此使用专用的内存数据库，例如 redis，它可以同时处理多个进程的轮询。

@Sam Mason 建议使用 WSL 和 fork 可能会更好，但这个是最便携的。

编辑：要将其存储在子全局范围内，您必须通过初始化程序传递它。

def define_global(var):
    global shared_dict
    shared_dict = var
...
if __name__ == "__main__":
...

    with multiprocessing.Pool(initializer=define_global, initargs=(shared_dict ,) as pool: