Python MongoDB (PyMongo) 多处理游标

Question

我正在尝试制作一个多处理 MongoDB 实用程序，它完美地工作，但我认为我有一个性能问题......即使有 20 个工人，它每秒处理的文档也不超过 2800 个......我想我可以得到快 5 倍...这是我的代码，它没有做任何异常的事情，只是将剩余时间打印到光标的末尾。

也许有更好的方法在 MongoDB 游标上执行多处理，因为我需要在每个包含 17.4M 记录集合的文档上运行一些东西，所以性能和更少的时间是必须的。

START = time.time()
def remaining_time(a, b):
    if START:
        y = (time.time() - START)
        z = ((a * y) / b) - y
        d = time.strftime('%H:%M:%S', time.gmtime(z))
        e = round(b / y)
        progress("{0}/{1} | Tiempo restante {2} ({3}p/s)".format(b, a, d, e), b, a)


def progress(p, c, t):
    pc = (c * 100) / t
    sys.stdout.write("%s [%-20s] %d%%\r" % (p, '?' * (pc / 5), pc))
    sys.stdout.flush()

def dowork(queue):
    for p, i, pcount in iter(queue.get, 'STOP'):
        remaining_time(pcount, i)


def populate_jobs(queue):
    mongo_query = {}
    products = MONGO.mydb.items.find(mongo_query, no_cursor_timeout=True)
    if products:
        pcount = products.count()
        i = 1
        print "Procesando %s productos..." % pcount
        for p in products:
            try:
                queue.put((p, i, pcount))
                i += 1
            except Exception, e:
                utils.log(e)
                continue
    queue.put('STOP')


def main():
    queue = multiprocessing.Queue()

    procs = [multiprocessing.Process(target=dowork, args=(queue,)) for _ in range(CONFIG_POOL_SIZE)]

    for p in procs:
        p.start()

    populate_jobs(queue)

    for p in procs:
        p.join()

另外，我注意到大约每 2500 个 aprox 文档，脚本会暂停大约 0.5 - 1 秒，这显然是一个坏问题。这是一个 MongoDB 问题，因为如果我执行完全相同的循环但使用range(0, 1000000)脚本根本不会暂停并且以每秒 57,000 次迭代运行，总共需要 20 秒来结束脚本......与 2,800 MongoDB 文档的巨大差异每秒...

这是运行 1,000,000 次迭代循环的代码，而不是 docs。

def populate_jobs(queue):
    mongo_query = {}
    products = MONGO.mydb.items.find(mongo_query, no_cursor_timeout=True)
    if products:
        pcount = 1000000
        i = 1
        print "Procesando %s productos..." % pcount
        for p in range(0, 1000000):
            queue.put((p, i, pcount))
            i += 1
    queue.put('STOP')

更新 正如我所看到的，问题不在于多处理本身，而是填充Queue未在多处理模式下运行的游标，这是一个填充Queue(populateJobs方法) 的简单过程，也许如果我可以使游标多线程/多进程并填充Queue同时，它会被更快地填充，然后多处理方法dowork会做得更快，因为我认为有一个瓶颈，我每秒只能填充大约 2,800 个项目Queue并在dowork多进程中检索更多，但我不知道我该怎么做并行化MongoDB游标。

也许，问题是我的计算机和服务器的 MongoDB 之间的延迟。在我要求下一个光标和 MongoDB 告诉我哪个是延迟之间，我的性能降低了 2000%（从 61,000 str/s 到 2,800 doc/s） NOPE我已经在本地主机 MongoDB 上尝试过，性能完全相同。 ..这让我发疯

Answer 1

以下是如何使用 aPool来喂养孩子：

START = time.time()
def remaining_time(a, b):
    if START:
        y = (time.time() - START)
        z = ((a * y) / b) - y
        d = time.strftime('%H:%M:%S', time.gmtime(z))
        e = round(b / y)
        progress("{0}/{1} | Tiempo restante {2} ({3}p/s)".format(b, a, d, e), b, a)


def progress(p, c, t):
    pc = (c * 100) / t
    sys.stdout.write("%s [%-20s] %d%%\r" % (p, '?' * (pc / 5), pc))
    sys.stdout.flush()

def dowork(args):
    p, i, pcount  = args
    remaining_time(pcount, i)

def main():
    queue = multiprocessing.Queue()

    procs = [multiprocessing.Process(target=dowork, args=(queue,)) for _ in range(CONFIG_POOL_SIZE)]
    pool = multiprocessing.Pool(CONFIG_POOL_SIZE)
    mongo_query = {}
    products = MONGO.mydb.items.find(mongo_query, no_cursor_timeout=True)
    pcount = products.count()
    pool.map(dowork, ((p, idx, pcount) for idx,p in enumerate(products)))
    pool.close()
    pool.join()

请注意， usingpool.map需要一次将所有内容从游标加载到内存中，但这可能是一个问题，因为它有多大。您可以使用imap来避免一次消耗整个内容，但您需要指定 achunksize以最小化 IPC 开销：

# Calculate chunksize using same algorithm used internally by pool.map
chunksize, extra = divmod(pcount, CONFIG_POOL_SIZE * 4)
if extra:
   chunksize += 1

pool.imap(dowork, ((p, idx, pcount) for idx,p in enumerate(products)), chunksize=chunksize)
pool.close()
pool.join()

对于 1,000,000 个项目，块大小为 12,500。您可以尝试更大或更小的尺寸，看看它如何影响性能。

我不确定这会有多大帮助，如果瓶颈实际上只是从 MongoDB 中提取数据。