在Python多进程中共享Pandas数据帧的字典

Question

我有一个Python Pandas数据框的字典，这个字典的总大小约为2GB。但是，当我在16个多进程之间共享它时（在子进程中，我只读取字典的数据而不修改它），它占用32GB的内存。所以我想问一下，在不复制它的情况下，是否可能跨多处理共享这个字典。我尝试将其转换为manager.dict()，但似乎耗费时间太长。最标准的方法是什么？谢谢。

- user40780

---

看看这个是否有帮助 - https://dev59.com/EWw15IYBdhLWcg3wGn5c - TrigonaMinima

+1 你可能也对这个问题感兴趣 share-a-dict-with-multiple-python-scripts 和那里的答案 when-is-copy-on-write-invoked-for-python-multiprocessing-across-class-methods。 - Darkonaut

你在寻找哪个操作系统的解决方案？ - Darkonaut

@Darkonaut，MAC OS和Linux - user40780

当你说它花费太长时间时，是指检索速度太慢，导致多进程必须等待每次检索，还是设置时间太长？我也很想知道你需要一次获取字典的多大比例（例如总大小的百分比），以及你需要多久抓取一次数据（几乎连续不断，还是在获取数据后有重要的处理过程）。 - bivouac0

Answer 1

我找到的最好的解决方案（它仅适用于某些类型的问题）是使用Python的BaseManager和SyncManager类来设置客户端/服务器。首先，您需要设置一个服务器，为数据提供代理类。

DataServer.py

#!/usr/bin/python
from    multiprocessing.managers import SyncManager
import  numpy

# Global for storing the data to be served
gData = {}

# Proxy class to be shared with different processes
# Don't put big data in here since that will force it to be piped to the
# other process when instantiated there, instead just return a portion of
# the global data when requested.
class DataProxy(object):
    def __init__(self):
        pass

    def getData(self, key, default=None):
        global gData
        return gData.get(key, None)

if __name__ == '__main__':
    port  = 5000

    print 'Simulate loading some data'
    for i in xrange(1000):
        gData[i] = numpy.random.rand(1000)

    # Start the server on address(host,port)
    print 'Serving data. Press <ctrl>-c to stop.'
    class myManager(SyncManager): pass
    myManager.register('DataProxy', DataProxy)
    mgr = myManager(address=('', port), authkey='DataProxy01')
    server = mgr.get_server()
    server.serve_forever()

运行上述代码一次并让它保持运行状态。下面是客户端类，您可以使用该类来访问数据。

DataClient.py

from   multiprocessing.managers import BaseManager
import psutil   #3rd party module for process info (not strictly required)

# Grab the shared proxy class.  All methods in that class will be availble here
class DataClient(object):
    def __init__(self, port):
        assert self._checkForProcess('DataServer.py'), 'Must have DataServer running'
        class myManager(BaseManager): pass
        myManager.register('DataProxy')
        self.mgr = myManager(address=('localhost', port), authkey='DataProxy01')
        self.mgr.connect()
        self.proxy = self.mgr.DataProxy()

    # Verify the server is running (not required)
    @staticmethod
    def _checkForProcess(name):
        for proc in psutil.process_iter():
            if proc.name() == name:
                return True
        return False

以下是使用多进程测试代码。

测试MP.py

#!/usr/bin/python
import time
import multiprocessing as mp
import numpy
from   DataClient import *    

# Confusing, but the "proxy" will be global to each subprocess, 
# it's not shared across all processes.
gProxy = None
gMode  = None
gDummy = None
def init(port, mode):
    global gProxy, gMode, gDummy
    gProxy  = DataClient(port).proxy
    gMode  = mode
    gDummy = numpy.random.rand(1000)  # Same as the dummy in the server
    #print 'Init proxy ', id(gProxy), 'in ', mp.current_process()

def worker(key):
    global gProxy, gMode, gDummy
    if 0 == gMode:   # get from proxy
        array = gProxy.getData(key)
    elif 1 == gMode: # bypass retrieve to test difference
        array = gDummy
    else: assert 0, 'unknown mode: %s' % gMode
    for i in range(1000):
        x = sum(array)
    return x    

if __name__ == '__main__':
    port   = 5000
    maxkey = 1000
    numpts = 100

    for mode in [1, 0]:
        for nprocs in [16, 1]:
            if 0==mode: print 'Using client/server and %d processes' % nprocs
            if 1==mode: print 'Using local data and %d processes' % nprocs                
            keys = [numpy.random.randint(0,maxkey) for k in xrange(numpts)]
            pool = mp.Pool(nprocs, initializer=init, initargs=(port,mode))
            start = time.time()
            ret_data = pool.map(worker, keys, chunksize=1)
            print '   took %4.3f seconds' % (time.time()-start)
            pool.close()

当我在我的电脑上运行这个时，我得到了...

Using local data and 16 processes
   took 0.695 seconds
Using local data and 1 processes
   took 5.849 seconds
Using client/server and 16 processes
   took 0.811 seconds
Using client/server and 1 processes
   took 5.956 seconds

无论在你的多进程系统中，这是否适用取决于你需要多频繁地获取数据。每次传输都会带来一些小的开销。如果你将x=sum(array)循环中的迭代次数降低，就可以看到这一点。在某个时候，你会花费更多的时间来获取数据而不是处理它。

除了多进程，我还喜欢这种模式，因为我只需在服务器程序中加载一次我的大型数组数据，并且它会一直保持加载状态，直到我关闭服务器。这意味着我可以针对数据运行一堆单独的脚本，并且它们执行得非常快；不需要等待数据加载。

虽然这里的方法有些类似于使用数据库，但它具有在任何类型的Python对象上工作的优势，而不仅仅是简单的字符串和整数等DB表格。我发现对于那些简单的类型使用数据库会更快一些，但对我来说，从编程上来讲，它往往更加困难，而且我的数据并不总是容易移植到数据库中。

- bivouac0

---

我应该指出，我已经在一个简单的库中实现了这个... https://github.com/bjascob/PythonDataServe - bivouac0