使用python多处理异步启动大量依赖进程

Question

问题：我有一个类似DAG（有向无环图）的结构，用于在机器上开始执行一些海量数据处理。某些流程只能在其父级数据处理完成时启动，因为存在多级处理。我想首先使用python多处理库在一台机器上处理它的所有内容，然后使用管理器在不同的机器上扩展执行。我以前没有python多处理的经验。有人能建议这是一个好的图书馆吗？如果是，一些基本的实现想法就可以了。如果不是，在python中还有什么可以用来做这件事？

例子：

A-

B-

C-

在上面的例子中我想踢A

附言：抱歉，我不能分享实际数据，因为保密，尽管我试图用这个例子来说明。

Answer 1

我非常喜欢为这样的事情使用进程和队列。

像这样：

from multiprocessing import Process, Queue
from Queue import Empty as QueueEmpty
import time

#example process functions
def processA(queueA, queueB):
    while True:
        try:
            data = queueA.get_nowait()
            if data == 'END':
                break
        except QueueEmpty:
            time.sleep(2) #wait some time for data to enter queue
            continue
        #do stuff with data
        queueB.put(data)

def processA(queueB, _):
    while True:
        try:
            data = queueB.get_nowait()
            if data == 'END':
                break
        except QueueEmpty:
            time.sleep(2) #wait some time for data to enter queue
            continue
        #do stuff with data

#helper functions for starting and stopping processes
def start_procs(num_workers, target_function, args):
    procs = []
    for _ in range(num_workers):
        p = Process(target=target_function, args=args)
        p.start()
        procs.append(p)
    return procs

def shutdown_process(proc_lst, queue):
    for _ in proc_lst:
        queue.put('END')
    for p in proc_lst:
        try:
            p.join()
        except KeyboardInterrupt:
            break

queueA = Queue(<size of queue> * 3) #needs to be a bit bigger than actual. 3x works well for me
queueB = Queue(<size of queue>)
queueC = Queue(<size of queue>)
queueD = Queue(<size of queue>)

procsA = start_procs(number_of_workers, processA, (queueA, queueB)) 
procsB = start_procs(number_of_workers, processB, (queueB, None))  

# feed some data to processA
[queueA.put(data) for data in start_data]  

#shutdown processes
shutdown_process(procsA, queueA)
shutdown_process(procsB, queueB)

#etc, etc. You could arrange the start, stop, and data feed statements to arrive at the dag behaviour you desire