在集群环境中用python设置管理器

Question

我希望您能就以下问题提供意见：

不久之后，我在一家新公司开始了一份工作，该公司在集群上运行进程。有一个已经实现的现有管道，大致执行以下操作：

1. 硬盘上每 +- 200 个大文件存储（每个文件 +- 130gb）
2. 由于集群上有磁盘配额，而且复制时 IO 非常密集， 我必须限制自己一次只能复制 1 个文件。
3. 管理器 Java 程序将创建一个拉取脚本，以通过网络（从 NAS 到集群）拉取大文件。
4. 拉取后，分析管道在集群上运行（对我来说是黑盒过程）。
5. 接下来，“我完成了吗”脚本正在检查集群上的进程是否已完成。如果不是，则脚本休眠 10 分钟并再次检查，如果完成，则删除大文件（对我来说是黑盒脚本）。

所以目前我已经在 python 中做了一个非常简单的管理程序实现，这样做，一旦完成，执行下一个文件以复制并重复作业列表。

我的问题是，我想扩展这个程序，这样它会使用 5 个（以后可能更多）大文件一次复制，并将它们提交到集群，并且只有在完成后才删除和删除一个进程正在运行。

在寻找解决方案时，我看到有人提到使用多线程或多处理，特别是使用工作池。我还没有这方面的经验（但是可以学习吗？），但我认为在这种情况下，这将是一个可行的选择。 我的问题是，我将如何设置一个由 5 个工作人员组成的池，以便每个工作人员执行一系列任务，并在完成后从队列中获取一个新的大文件并进行迭代。

Answer 1

multiprocessing.Pool 专为这个确切的用例而设计：

import multiprocessing

def process_big_file(big_file):
    print("Process {0}: Got big file {1}".format(multiprocessing.current_process(), big_file))
    return "done with {0}".format(big_file)

def get_big_file_list():
    return ['bf{0}'.format(i) for i in range(20)]  # Just a dummy list


if __name__ == "__main__":
    pool = multiprocessing.Pool(5)  # 5 worker processes in the pool
    big_file_list = get_big_file_list()
    results = pool.map(process_big_file, big_file_list)
    print(results)

输出：

Process <Process(PoolWorker-1, started daemon)>: Got big file bf0
Process <Process(PoolWorker-1, started daemon)>: Got big file bf1
Process <Process(PoolWorker-3, started daemon)>: Got big file bf2
Process <Process(PoolWorker-4, started daemon)>: Got big file bf3
Process <Process(PoolWorker-5, started daemon)>: Got big file bf4
Process <Process(PoolWorker-5, started daemon)>: Got big file bf5
Process <Process(PoolWorker-5, started daemon)>: Got big file bf6
Process <Process(PoolWorker-3, started daemon)>: Got big file bf7
Process <Process(PoolWorker-3, started daemon)>: Got big file bf8
Process <Process(PoolWorker-2, started daemon)>: Got big file bf9
Process <Process(PoolWorker-2, started daemon)>: Got big file bf10
Process <Process(PoolWorker-2, started daemon)>: Got big file bf11
Process <Process(PoolWorker-4, started daemon)>: Got big file bf12
Process <Process(PoolWorker-4, started daemon)>: Got big file bf13
Process <Process(PoolWorker-4, started daemon)>: Got big file bf14
Process <Process(PoolWorker-4, started daemon)>: Got big file bf15
Process <Process(PoolWorker-4, started daemon)>: Got big file bf16
Process <Process(PoolWorker-4, started daemon)>: Got big file bf17
Process <Process(PoolWorker-4, started daemon)>: Got big file bf18
Process <Process(PoolWorker-4, started daemon)>: Got big file bf19

['done with bf0', 'done with bf1', 'done with bf2', 'done with bf3', 'done with bf4', 'done with bf5', 'done with bf6', 'done with bf7', 'done with bf8', 'done with bf9', 'done with bf10', 'done with bf11', 'done with bf12', 'done with bf13', 'done with bf14', 'done with bf15', 'done with bf16', 'done with bf17', 'done with bf18', 'done with bf19']

pool.map 调用使用内部队列将big_file_list 中的所有项目分发给队列中的工作人员。一旦工作人员完成一项任务，它就会从队列中拉出下一个项目，并一直持续到队列为空。