在 python 3.4.7 中的 pool.map 函数中添加额外的随机参数作为参数

Question

我想在大型数据集上使用多处理来查找两列的乘积并使用参数中的给定参数过滤数据集。我构建了一个测试集，但我一直无法让多处理在这个集上工作。

首先，我试图在 parallelize_dataframe 函数中划分数据集，然后在 subset_col 函数中应用乘法函数和过滤函数。稍后我将完整的数据集追加到 parallelize_dataframe.

import numpy as np
import pandas as pd
from multiprocessing import Pool
from multiprocessing import Lock

df = pd.DataFrame({'col1': [1, 0, 1, 1, 1, 0, 0, 1, 0, 1],
                'col2': ['aa', 'aa', 'aa', 'bb', 'bb', 'bb', 'bb', 'cc', 'cc', 'cc'],
                'col3': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
                'col4': [21, 22, 23, 24, 25, 26, 27, 28, 29, 30]})



def subset_col(df, p):
    print("Working with number: " + str(p))
    df[col5] = df[col3]*df[col4]
    df= df[df['col1'] == p]


def parallelize_dataframe(df, p, func, n_cores=80):
    df_split = np.array_split(df, n_cores)
    pool = Pool(n_cores)
    df = pd.concat(pool.map(func, df_split, p))
    pool.close()
    pool.join()
    return df


df3 = parallelize_dataframe(df,1,subset_col)

结果应该是 col3 和 col4 的乘积，col1 用值过滤。但我总是收到一条错误消息：

File "<stdin>", line 1, in <module>
File "<stdin>", line 4, in parallelize_dataframe
struct.error: 'i' format requires -2147483648 <= number <= 2147483647

但是，如果我从所有函数中删除过滤器 "p"，它就完全可以正常工作。有人可以帮我调试吗？

Answer 1

来自multiprocessing.Pool.map的官方文档，"supports supports only one iterable argument"。因此，您需要更改 subset_col 的接口以改为采用单个参数。此外，您忘记将列设为字符串，从而导致名称错误。为了减少计算量，您应该在乘法之前进行过滤。然后应该返回一个值，除非您的函数仅通过副作用运行（我假设您不希望这样做，因为您连接了池结果）。

def subset_col(pair):
    df, p = pair
    print("Working with number: " + str(p))
    df = df[df['col1'] == p].copy()
    df['col5'] = df['col3']
    return df

接下来，我们将需要修复您调用 pool.map 的方式，因为根据您的操作，它应该只需要 2 个参数（第三个，最后一个参数是 chunksize）。由于您希望对每个进程使用相同的 p，因此我们会将 dfs 与每个进程的重复值 p 压缩在一起。另外，考虑使用上下文管理器来处理关闭资源。

def parallelize_dataframe(df, p, func, n_cores=None):
    if n_cores is None:
        n_cores = os.cpu_count()

    dfs = np.array_split(df, n_cores)
    pairs = zip(dfs, itertools.repeat(p))
    with Pool(n_cores) as pool:
        result = pool.map(func, pairs)

    df = pd.concat(result)
    return df

这现在是 returns 新的数据框。但我绝对怀疑你有一台 80 核的机器。考虑使用 os.cpu_count

实施 n_cores=None 让 Python dynamically figure out 你的机器上有多少核心

df3 = parallelize_dataframe(df, 1, subset_col)

根据您对 Pool.starmap 变体的要求：

def subset_col(df, p):
    # remove unpacking line
    ...

def parallelize_dataframe(df, p, func, n_cores=None):
    ...
    # change `pool.map(...)` to `pool.starmap(...)`
    ...

但是您应该注意，Pool 不提供 imap 或 imap_unordered 替代 starmap，它们都是惰性评估版本，不同的是是否是否保留顺序。

在 python 3.4.7 中的 pool.map 函数中添加额外的随机参数作为参数

Adding additional random parameter as an argument in pool.map function in python 3.4.7

python

numpy

multiprocess

pandas

python-multiprocessing