具有joblib并行的Python

Question

我有两个类，第一个类是我的基本实现，第二个类使用类一个并使用joblib的并行调用它的方法，这样就可以这样：

class MyClass:
    def __init__(self, param_list: list):
        self.param_list = param_list

    def method1(self, w, z):
        pass

    def method2(self, w, y, z):
        pass

现在，第二个类实现如下：

from joblib import Parallel, delayed

class MyParallelClass:
    def __init__(self, param_list: list[dict]):
        self.param_list = param_list

    def method1(self, x: list[dict]):
        results = Parallel(n_jobs=-1)(delayed(MyClass(**params).method1)(**arguments)
                                      for params in self.param_list
                                      for arguments in x)
        return results

    def method2(self, x: list[dict]):
        results = Parallel(n_jobs=-1)(delayed(MyClass(**params).method2)(**arguments)
                                      for params in self.param_list
                                      for arguments in x)
        return results

这是可行的，但如您所见，MyParallelClass方法1和方法2本质上是相同的，唯一更改的是从MyClass调用方法。

因此，我想简化这个问题，有一个处理并行执行的内部方法，给出一个参数，该参数告诉我要执行哪种方法，所以如下所示：

class MyParallelClass:
    def __init__(self, param_list: list[dict]):
        self.param_list = param_list

    def method1(self, x: list[dict]):
        return self._run(x, method="method1")

    def method2(self, x: list[dict]):
        return self._run(x, method="method2")
    
    def _run(self, x, method):
        pass

如何编写此_run方法？我试过像这样使用标准getatrr，但失败了：

def _run(self, x: list[dict], method):
    results = Parallel(n_jobs=-1)(delayed(getattr(MyClass(**params), method))(**arguments)
                                  for params in self.param_list
                                  for arguments in x)
    return results 

但我知道这个错误：

TypeError: cannot unpack non-iterable function object

Answer 1

这个例子解释了如何重用类方法吗？

class Base:

  def __init__(self, init_arg):
    self.init_arg = init_arg

  def method1(self, x):
    print(x, "from method1,", "iteration_number is", self.init_arg)
    return f"lower value is {x.lower()} [iteration number is {self.init_arg}]"

  def method2(self, x, y):
    print(x, y, "from method2,", "iteration_number is", self.init_arg)
    return f"lower values are {x.lower()}, {y.lower()} [iteration number is {self.init_arg}]"

class Executor:

  def __init__(self, init_arg):
    self.base_instance = Base(init_arg)

  def _method_base(self, method_name, *args, **kwargs):
    return getattr(self.base_instance, method_name)(*args, **kwargs)

  def method1(self, x):
    return self._method_base('method1', x)

  # here x is *args and y=y is **kwargs
  def method2(self, x, y):
    return self._method_base('method2', x, y=y)

for idx, val in enumerate([("X1", "Y1"), ("X2", "Y2"), ("X3", "Y3")]):
  executor = Executor(idx)
  _1 = executor.method1(val[0])
  _2 = executor.method2(val[0], val[1])
  print("calculated values: ", _1, _2)

输出：

X1 from method1, iteration_number is 0
X1 Y1 from method2, iteration_number is 0
calculated values:  lower value is x1 [iteration number is 0] lower values are x1, y1 [iteration number is 0]
X2 from method1, iteration_number is 1
X2 Y2 from method2, iteration_number is 1
calculated values:  lower value is x2 [iteration number is 1] lower values are x2, y2 [iteration number is 1]
X3 from method1, iteration_number is 2
X3 Y3 from method2, iteration_number is 2
calculated values:  lower value is x3 [iteration number is 2] lower values are x3, y3 [iteration number is 2]