- mongodb - 在 MongoDB mapreduce 中,如何展平值对象?
- javascript - 对象传播与 Object.assign
- html - 输入类型 ="submit"Vs 按钮标签它们可以互换吗?
- sql - 使用 MongoDB 而不是 MS SQL Server 的优缺点
在多处理模块中使用 Pool.map_async()
(以及 Pool.map()
)时遇到问题。我已经实现了一个并行循环函数,只要 Pool.map_async
的函数输入是“常规”函数,它就可以正常工作。当功能是例如一个类的方法,然后我得到一个 PicklingError
:
cPickle.PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed
我只使用 Python 进行科学计算,所以我对酸洗的概念不是很熟悉,今天刚刚了解了一点。我看过几个以前的答案,比如 Can't pickle <type 'instancemethod'> when using multiprocessing Pool.map() ,但我无法弄清楚如何使它工作,即使按照答案中提供的链接进行操作。
我的代码,其目标是使用多核模拟 Normal r.v 的向量。请注意,这只是一个示例,甚至可能无法在多核上运行。
import multiprocessing as mp
import scipy as sp
import scipy.stats as spstat
def parfor(func, args, static_arg = None, nWorkers = 8, chunksize = None):
"""
Purpose: Evaluate function using Multiple cores.
Input:
func - Function to evaluate in parallel
arg - Array of arguments to evaluate func(arg)
static_arg - The "static" argument (if any), i.e. the variables that are constant in the evaluation of func.
nWorkers - Number of Workers to process computations.
Output:
func(i, static_arg) for i in args.
"""
# Prepare arguments for func: Collect arguments with static argument (if any)
if static_arg != None:
arguments = [[arg] + static_arg for arg in list(args)]
else:
arguments = args
# Initialize workers
pool = mp.Pool(processes = nWorkers)
# Evaluate function
result = pool.map_async(func, arguments, chunksize = chunksize)
pool.close()
pool.join()
return sp.array(result.get()).flatten()
# First test-function. Freeze location and scale for the Normal random variates generator.
# This returns a function that is a method of the class Norm_gen. Methods cannot be pickled
# so this will give an error.
def genNorm(loc, scale):
def subfunc(a):
return spstat.norm.rvs(loc = loc, scale = scale, size = a)
return subfunc
# Second test-function. The same as above but does not return a method of a class. This is a "plain" function and can be
# pickled
def test(fargs):
x, a, b = fargs
return spstat.norm.rvs(size = x, loc = a, scale = b)
# Try it out.
N = 1000000
# Set arguments to function. args1 = [1, 1, 1,... ,1], the purpose is just to generate a random variable of size 1 for each
# element in the output vector.
args1 = sp.ones(N)
static_arg = [0, 1] # standarized normal.
# This gives the PicklingError
func = genNorm(*static_arg)
sim = parfor(func, args1, static_arg = None, nWorkers = 12, chunksize = None)
# This is OK:
func = test
sim = parfor(func, args1, static_arg = static_arg, nWorkers = 12, chunksize = None)
点击 Can't pickle <type 'instancemethod'> when using multiprocessing Pool.map() 中问题答案中提供的链接, Steven Bethard(几乎在最后)建议使用 copy_reg
模块。他的代码是:
def _pickle_method(method):
func_name = method.im_func.__name__
obj = method.im_self
cls = method.im_class
return _unpickle_method, (func_name, obj, cls)
def _unpickle_method(func_name, obj, cls):
for cls in cls.mro():
try:
func = cls.__dict__[func_name]
except KeyError:
pass
else:
break
return func.__get__(obj, cls)
import copy_reg
import types
copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)
我真的不明白如何使用它。我唯一能想到的就是把它放在我的代码之前,但它没有帮助。一个简单的解决方案当然是只使用可行的解决方案,避免涉及 copy_reg
。我更感兴趣的是让 copy_reg
正常工作以充分利用多处理,而不必每次都解决问题。
最佳答案
这里的问题不是“泡菜”错误消息而是概念性的:多进程确实在“ worker ”不同的进程中 fork 你的代码以执行它的魔力。
然后,它通过无缝序列化和反序列化数据(即使用 pickle 的部分)将数据发送到不同的进程或从不同的进程发送数据。
当来回传递的部分数据是一个函数时 - 它假设被调用进程中存在同名的函数,并且(我猜)将函数名作为字符串传递。由于函数是无状态的,被调用的工作进程只是用它收到的数据调用同一个函数。(Python函数不能通过pickle进行序列化,所以只在主进程和工作进程之间传递引用)
当你的函数是实例中的一个方法时——尽管当我们编写 python 代码时,它与函数很像,带有一个“自动”self
变量,但在下面就不一样了.因为实例(对象)是有状态的。这意味着工作进程没有对象的副本,该对象是您要在另一端调用的方法的所有者。
解决将方法作为函数传递给 map_async 调用的方法也不起作用 - 因为多进程只使用函数引用,而不是传递它时的实际函数。
因此,您应该 (1) 更改您的代码,以便将函数(而不是方法)传递给工作进程,将对象保持的任何状态转换为要调用的新参数。(2) 为 map_async 调用创建一个“目标”函数,该函数在工作进程端重建所需的对象,然后调用其中的函数。 Python 中最直接的类本身是可挑选的,因此您可以在 map_async 调用中传递作为函数所有者本身的对象 - “目标”函数将在工作端调用适当的方法本身。
(2) 可能听起来“困难”,但可能就是这样——除非你的对象的类不能被腌制:
import types
def target(object, *args, **kw):
method_name = args[0]
return getattr(object, method_name)(*args[1:])
(...)
#And add these 3 lines prior to your map_async call:
# Evaluate function
if isinstance (func, types.MethodType):
arguments.insert(0, func.__name__)
func = target
result = pool.map_async(func, arguments, chunksize = chunksize)
*免责声明:我没有测试过这个
关于python - 使用多处理时出现 PicklingError,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/7016567/
SQLite、Content provider 和 Shared Preference 之间的所有已知区别。 但我想知道什么时候需要根据情况使用 SQLite 或 Content Provider 或
警告:我正在使用一个我无法完全控制的后端,所以我正在努力解决 Backbone 中的一些注意事项,这些注意事项可能在其他地方更好地解决......不幸的是,我别无选择,只能在这里处理它们! 所以,我的
我一整天都在挣扎。我的预输入搜索表达式与远程 json 数据完美配合。但是当我尝试使用相同的 json 数据作为预取数据时,建议为空。点击第一个标志后,我收到预定义消息“无法找到任何内容...”,结果
我正在制作一个模拟 NHL 选秀彩票的程序,其中屏幕右侧应该有一个 JTextField,并且在左侧绘制弹跳的选秀球。我创建了一个名为 Ball 的类,它实现了 Runnable,并在我的主 Draf
这个问题已经有答案了: How can I calculate a time span in Java and format the output? (18 个回答) 已关闭 9 年前。 这是我的代码
我有一个 ASP.NET Web API 应用程序在我的本地 IIS 实例上运行。 Web 应用程序配置有 CORS。我调用的 Web API 方法类似于: [POST("/API/{foo}/{ba
我将用户输入的时间和日期作为: DatePicker dp = (DatePicker) findViewById(R.id.datePicker); TimePicker tp = (TimePic
放宽“邻居”的标准是否足够,或者是否有其他标准行动可以采取? 最佳答案 如果所有相邻解决方案都是 Tabu,则听起来您的 Tabu 列表的大小太长或您的释放策略太严格。一个好的 Tabu 列表长度是
我正在阅读来自 cppreference 的代码示例: #include #include #include #include template void print_queue(T& q)
我快疯了,我试图理解工具提示的行为,但没有成功。 1. 第一个问题是当我尝试通过插件(按钮 1)在点击事件中使用它时 -> 如果您转到 Fiddle,您会在“内容”内看到该函数' 每次点击都会调用该属
我在功能组件中有以下代码: const [ folder, setFolder ] = useState([]); const folderData = useContext(FolderContex
我在使用预签名网址和 AFNetworking 3.0 从 S3 获取图像时遇到问题。我可以使用 NSMutableURLRequest 和 NSURLSession 获取图像,但是当我使用 AFHT
我正在使用 Oracle ojdbc 12 和 Java 8 处理 Oracle UCP 管理器的问题。当 UCP 池启动失败时,我希望关闭它创建的连接。 当池初始化期间遇到 ORA-02391:超过
关闭。此题需要details or clarity 。目前不接受答案。 想要改进这个问题吗?通过 editing this post 添加详细信息并澄清问题. 已关闭 9 年前。 Improve
引用这个plunker: https://plnkr.co/edit/GWsbdDWVvBYNMqyxzlLY?p=preview 我在 styles.css 文件和 src/app.ts 文件中指定
为什么我的条形这么细?我尝试将宽度设置为 1,它们变得非常厚。我不知道还能尝试什么。默认厚度为 0.8,这是应该的样子吗? import matplotlib.pyplot as plt import
当我编写时,查询按预期执行: SELECT id, day2.count - day1.count AS diff FROM day1 NATURAL JOIN day2; 但我真正想要的是右连接。当
我有以下时间数据: 0 08/01/16 13:07:46,335437 1 18/02/16 08:40:40,565575 2 14/01/16 22:2
一些背景知识 -我的 NodeJS 服务器在端口 3001 上运行,我的 React 应用程序在端口 3000 上运行。我在 React 应用程序 package.json 中设置了一个代理来代理对端
我面临着一个愚蠢的问题。我试图在我的 Angular 应用程序中延迟加载我的图像,我已经尝试过这个2: 但是他们都设置了 src attr 而不是 data-src,我在这里遗漏了什么吗?保留 d
我是一名优秀的程序员,十分优秀!