首页
学习
活动
专区
工具
TVP
发布
社区首页 >问答首页 >python有没有一个简单的基于进程的并行映射?

python有没有一个简单的基于进程的并行映射?
EN

Stack Overflow用户
提问于 2009-11-10 06:33:42
回答 5查看 49.6K关注 0票数 75

我正在寻找一个简单的基于进程的python并行映射,即一个函数

代码语言:javascript
复制
parmap(function,[data])

这将在不同进程上的每个数据元素上运行函数(嗯,在不同的内核上,但AFAIK,在python中在不同内核上运行的唯一方法是启动多个解释器),并返回结果列表。

真的存在这样的东西吗?我想一些简单的的东西,所以一个简单的模块会很好。当然,如果不存在这样的东西,我会满足于一个大的库:-/

EN

回答 5

Stack Overflow用户

回答已采纳

发布于 2009-11-10 06:49:56

我觉得你需要的是map method in multiprocessing.Pool()

映射(函数,可迭代,块大小)

一个与map()内置函数并行的等价物(尽管它只支持一个可迭代参数)。它会一直阻塞,直到结果准备就绪。此方法将迭代器分成多个块,并将这些块作为单独的任务提交给进程池。这些块的(近似)大小可以通过将chunksize设置为正整数来指定

例如,如果您想映射此函数:

代码语言:javascript
复制
def f(x):
    return x**2

要使用range(10),可以使用内置的map()函数:

代码语言:javascript
复制
map(f, range(10))

或者使用multiprocessing.Pool()对象的方法map():

代码语言:javascript
复制
import multiprocessing
pool = multiprocessing.Pool()
print pool.map(f, range(10))
票数 136
EN

Stack Overflow用户

发布于 2019-02-07 07:14:47

使用Ray可以很好地做到这一点,该系统允许您轻松地并行化和分发您的Python代码。

要使示例并行化,需要用@ray.remote装饰器定义映射函数,然后用.remote调用它。这将确保远程函数的每个实例都将在不同的进程中执行。

代码语言:javascript
复制
import time
import ray

ray.init()

# Define the function you want to apply map on, as remote function. 
@ray.remote
def f(x):
    # Do some work...
    time.sleep(1)
    return x*x

# Define a helper parmap(f, list) function.
# This function executes a copy of f() on each element in "list".
# Each copy of f() runs in a different process.
# Note f.remote(x) returns a future of its result (i.e., 
# an identifier of the result) rather than the result itself.  
def parmap(f, list):
    return [f.remote(x) for x in list]

# Call parmap() on a list consisting of first 5 integers.
result_ids = parmap(f, range(1, 6))

# Get the results
results = ray.get(result_ids)
print(results)

这将打印以下内容:

代码语言:javascript
复制
[1, 4, 9, 16, 25]

并且它将以大约len(list)/p (向上舍入最接近的整数)结束,其中p是您的机器上的核心数量。假设一台机器有两个核心,我们的示例将以5/2向上舍入执行,即在大约3秒内执行。

multiprocessing模块相比,使用Ray有许多优点。具体地说,相同的代码既可以在一台机器上运行,也可以在机器集群上运行。有关Ray的更多优势,请参阅this related post

票数 8
EN

Stack Overflow用户

发布于 2019-01-22 04:48:54

对于那些希望Python等同于R的mclapply()的人,这里是我的实现。它是对以下两个示例的改进:

@Rafael Valero.

提到的

它可以应用于具有单个或多个参数映射函数。

代码语言:javascript
复制
import numpy as np, pandas as pd
from scipy import sparse
import functools, multiprocessing
from multiprocessing import Pool

num_cores = multiprocessing.cpu_count()

def parallelize_dataframe(df, func, U=None, V=None):

    #blockSize = 5000
    num_partitions = 5 # int( np.ceil(df.shape[0]*(1.0/blockSize)) )
    blocks = np.array_split(df, num_partitions)

    pool = Pool(num_cores)
    if V is not None and U is not None:
        # apply func with multiple arguments to dataframe (i.e. involves multiple columns)
        df = pd.concat(pool.map(functools.partial(func, U=U, V=V), blocks))
    else:
        # apply func with one argument to dataframe (i.e. involves single column)
        df = pd.concat(pool.map(func, blocks))

    pool.close()
    pool.join()

    return df

def square(x):
    return x**2

def test_func(data):
    print("Process working on: ", data.shape)
    data["squareV"] = data["testV"].apply(square)
    return data

def vecProd(row, U, V):
    return np.sum( np.multiply(U[int(row["obsI"]),:], V[int(row["obsJ"]),:]) )

def mProd_func(data, U, V):
    data["predV"] = data.apply( lambda row: vecProd(row, U, V), axis=1 )
    return data

def generate_simulated_data():

    N, D, nnz, K = [302, 184, 5000, 5]
    I = np.random.choice(N, size=nnz, replace=True)
    J = np.random.choice(D, size=nnz, replace=True)
    vals = np.random.sample(nnz)

    sparseY = sparse.csc_matrix((vals, (I, J)), shape=[N, D])

    # Generate parameters U and V which could be used to reconstruct the matrix Y
    U = np.random.sample(N*K).reshape([N,K])
    V = np.random.sample(D*K).reshape([D,K])

    return sparseY, U, V

def main():
    Y, U, V = generate_simulated_data()

    # find row, column indices and obvseved values for sparse matrix Y
    (testI, testJ, testV) = sparse.find(Y)

    colNames = ["obsI", "obsJ", "testV", "predV", "squareV"]
    dtypes = {"obsI":int, "obsJ":int, "testV":float, "predV":float, "squareV": float}

    obsValDF = pd.DataFrame(np.zeros((len(testV), len(colNames))), columns=colNames)
    obsValDF["obsI"] = testI
    obsValDF["obsJ"] = testJ
    obsValDF["testV"] = testV
    obsValDF = obsValDF.astype(dtype=dtypes)

    print("Y.shape: {!s}, #obsVals: {}, obsValDF.shape: {!s}".format(Y.shape, len(testV), obsValDF.shape))

    # calculate the square of testVals    
    obsValDF = parallelize_dataframe(obsValDF, test_func)

    # reconstruct prediction of testVals using parameters U and V
    obsValDF = parallelize_dataframe(obsValDF, mProd_func, U, V)

    print("obsValDF.shape after reconstruction: {!s}".format(obsValDF.shape))
    print("First 5 elements of obsValDF:\n", obsValDF.iloc[:5,:])

if __name__ == '__main__':
    main()
票数 6
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/1704401

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档