平行的Python迭代[英] Parallel python iteration

本文是小编为大家收集整理的关于平行的Python迭代的处理方法,想解了平行的Python迭代的问题怎么解决?平行的Python迭代问题的解决办法?那么可以参考本文帮助大家快速定位并解决问题。

问题描述

我想根据 pandas.DataFrame 中的值创建一个类的多个实例.这个我记下来了.

import itertools
import multiprocessing as mp
import pandas as pd

class Toy:
    id_iter = itertools.count(1)

    def __init__(self, row):
        self.id = self.id_iter.next()
        self.type = row['type']

if __name__ == "__main__":

    table = pd.DataFrame({
        'type': ['a', 'b', 'c'],
        'number': [5000, 4000, 30000]
        })

    for index, row in table.iterrows():
        [Toy(row) for _ in range(row['number'])]

多处理尝试

我已经能够通过添加以下内容(在某种程度上)并行化:

pool = mp.Pool(processes=mp.cpu_count())
m = mp.Manager()
q = m.Queue()

for index, row in table.iterrows():
    pool.apply_async([Toy(row) for _ in range(row['number'])])

如果 row['number'] 中的数字比 table 的长度长得多,这似乎会更快.但在我的实际情况中,table 有数千行,每个 row['number'] 都比较小.

尝试将 table 分解为 cpu_count() 块并在表中迭代似乎更聪明.但现在我们正处于我的 Python 技能的边缘.

我已经尝试过 python 解释器对我大喊大叫的事情,例如:

pool.apply_async(
        for index, row in table.iterrows(): 
        [Toy(row) for _ in range(row['number'])]
        )

还有"不能腌制"的东西

Parallel(n_jobs=4)(
    delayed(Toy)([row for _ in range(row['number'])]) \
            for index, row in table.iterrows()
)

编辑

这可能让我更接近一点,但仍然没有.我在单独的函数中创建类实例,

def create_toys(row):
    [Toy(row) for _ in range(row['number'])]

....

Parallel(n_jobs=4, backend="threading")(
    (create_toys)(row) for i, row in table.iterrows()
)

但有人告诉我"NoneType"对象不可迭代.

推荐答案

我有点不清楚你期望的输出是什么.你只是想要一个表格的大列表

[Toy(row_1) ... Toy(row_n)]

每个 Toy(row_i) 以多重性出现在哪里 row_i.number?

根据@JD Long 提到的答案,我认为您可以这样做:

def process(df):
    L = []
    for index, row in table.iterrows():
        L += [Toy(row) for _ in range(row['number'])]
    return L

table = pd.DataFrame({
    'type': ['a', 'b', 'c']*10,
    'number': [5000, 4000, 30000]*10
    })

p = mp.Pool(processes=8)
split_dfs = np.array_split(table,8)    
pool_results = p.map(process, split_dfs)
p.close()
p.join()

# merging parts processed by different processes
result = [a for L in pool_results for a in L]

本文地址:https://www.itbaoku.cn/post/1727873.html