python程序并行处理出错

Question

def bagging_and_trees_growth(samples, network, tree_num):
    trees = []
    
    for i in range(tree_num):
        bootstrap_samples = bagging(samples)
        a_tree = tree_growth(network, bootstrap_samples)
        trees.append(a_tree)
        
    return trees
    
def agiled_random_forest(samples, network, size, processes=39):
   
    rforest = []
        
    #job_server = pp.Server(processes=processes)
    threadPool = ThreadPool(processes=processes)
     
    depfun = (find_best_split, stopping_condition, purity_gain, Gini_index, find_neighbors, tree_growth, bagging)
    dep_modules = ('networkx', 'numpy', 'math', 'random', 'sys', 'pNGF')
    
    
    tree_num_of_each_task = int(size/processes)
    
    #jobs = [pp_server.submit(bagging_and_trees_growth, (samples, network, tree_num_of_each_task), depfun, 'dep_modules) for x in range(processes)]
   
    jobs = [threadPool.apply_async(bagging_and_trees_growth, (samples, network, tree_num_of_each_task), depfun, dep_modules) for x in range(processes)]
    
    
    for job in jobs:
        rforest += job.get()
    
    threadPool.destroy()
    return rforest

它显示了映射和元组的错误

TypeError: bagging_and_trees_growth() argument after ** must be a mapping, not tuple

如何解决这个错误，因为pp鼠标在python3中无法工作？

Answer 1

您可能正在寻找这样的东西。

这里的想法是，bagging_and_trees_growth不再有一个内部作业循环；我们依靠线程池(或者，最好是GIL考虑的进程池，但这取决于您)来高效地处理分发工作。

由于作业的执行顺序在这里显然没有区别，因此imap_unordered将是最快的高级构造。一个人也可以使用apply_async，但它的工作量更大。

import itertools
import multiprocessing.pool


def bagging_and_trees_growth(job):
    samples, network = job  # unpack the job tuple
    bootstrap_samples = bagging(samples)
    a_tree = tree_growth(network, bootstrap_samples)
    return a_tree


def agiled_random_forest(samples, network, size, processes=39):
    rforest = []
    with multiprocessing.pool.ThreadPool(processes=processes) as pool:
        # to use imap_unordered (the fastest high-level pool operation),
        # we need to pack each job into an object; since all we need here is 2 parameters, let's use a tuple.
        # set up a generator to generate the same job size times
        job_gen = itertools.repeat((samples, network), size)
        # do the work in parallel
        for result in pool.imap_unordered(bagging_and_trees_growth, job_gen):
            # could do something else with the result here;
            # in fact this could all just be `rforest = list(pool.imap...)`
            # in the simple case
            rforest.append(result)
    return rforest