使用 Python 的并行编程工具处理数据文件答案

【问题标题】：Using Python's parallel programming facilities to process data files使用 Python 的并行编程工具处理数据文件
【发布时间】：2017-09-08 19:43:57
【问题描述】：

我正在尝试使用 python 通过程序并行运行文件列表。

为了抽象出我这样做的真正原因，我编写了一个名为fake_program.py 的小型命令行程序，它接受一个包含用逗号分隔的任意数字列表作为参数的文本文件。程序将数字相加并打印结果。

fake_program.py:

import argparse
##parse arguments from command line 
parser = argparse.ArgumentParser()
parser.add_argument('filename', type=str)
args = parser.parse_args()

## read data file
with open(args.filename) as f:
    numbers = f.read()

## process and add the numbers 
numbers = numbers.split(',')
numbers = [int(i) for i in numbers]
print reduce(lambda x, y: x + y, numbers)

示例输入文件：

data_file_1.txt 包含：

1, 2, 3

data_file2.txt 包含

4, 5, 6

data_file3.txt 包含

7, 8, 9

有用法：

$ python fake_program.py data_file1.txt

output:

我的目标是通过fake_program.py 并行处理这三个文件（或任意数量的类似文件）。请注意，对于这个简单的示例，只需运行 fake_program.py 的单独实例即可处理所有数据。但是，我真正运行的程序计算量很大，并且在大约 8 个实例时变得无响应。由于我通过该程序运行了大约 400 个数据文件，并且每个数据文件最多需要 5 分钟来处理，因此最好引入一个Queue，这样（比如说）可以一次处理 5 个数据文件，当一个完成时另一个开始。

这是目前为止的脚本：

from subprocess import Popen, PIPE
from threading import Thread
import Queue
from multiprocessing import cpu_count
import glob

def run1(q, fle):
    """
    Process 1 file 
    :return: 
    """
    args = ['python', 'fake_program.py', '"{}"'.format(fle)]
    proc = Popen(args, stdin=PIPE, stdout=PIPE,
                 stderr=PIPE, shell=False)

    q.put(proc, block=True)



def run_all(files):
    """

    """
    number_of_processes = cpu_count()
    q = Queue.Queue(number_of_processes)
    result = []
    for fle in files:
        t = Thread(target=run1, args=(q, fle))
        t.start()
        q.join()
        result.append(q.get())
    return result   

if __name__=='__main__':
    working_directory = </my/working/directory>
    data_files = glob.glob(working_directory+'\*.txt')
    print data_files
    print run_all(data_files)

这个程序目前只是无限挂起。有人对我如何实现这种并行性有任何建议吗？

【问题讨论】：

最简单的方法是拆分数据并运行程序的三个实例，您可以使用简短的 shell 脚本轻松编排。
是的，但这是对我正在处理的另一个问题的抽象 - fake_program 实际上是一个 c++ 命令行程序，可以完成繁重的工作。因此我*必须以这种方式运行程序。感谢您的回复。
也许您应该描述您的实际问题而不是“C++ 中另一个问题的抽象”？大概你想要一个解决方案而不是一个解决方案的抽象。
...我有。我正在尝试在 python 而不是 C++ 中执行此操作。只是我正在使用的第 3 方程序是 c++，即 fake_program.py 确实是一个可执行文件（抱歉误导）
对，但这并不能解释为什么您不能只启动程序的三个实例。使用 python 合并结果，如果必须，使用 shell 来编排流程。它很擅长。

标签： python multithreading parallel-processing

【解决方案1】：

也许使用 multiprocessing.Pool 而不是 Queue。如果您的作业是异步的，那么使用起来非常简单。此外，如果您的子进程抛出异常，Python 多处理可能不够聪明，无法将其返回到主进程。我以前见过这种情况下挂起。

【讨论】：