python - 在 spark 上使用 python lime 作为 udf

Question

我正在寻找使用 lime's explainer 在 pyspark 的 udf 中。我以前训练过表格解释器，并按照 link 中的建议将其存储为 dill 模型

loaded_explainer = dill.load(open('location_to_explainer','rb'))

def lime_explainer(*cols):
    selected_cols = np.array([value for value in cols])
    exp = loaded_explainer.explain_instance(selected_cols, loaded_model.predict_proba, num_features = 10)
    mapping = exp.as_map()[1]

    return str(mapping)

然而，这会花费很多时间，因为看起来很多计算都是在驱动程序上进行的。然后我一直在尝试使用 spark broadcast 将解释器广播给执行者。

broadcasted_explainer= sc.broadcast(loaded_explainer)

def lime_explainer(*col):
    selected_cols = np.array([value for value in cols])
    exp = broadcasted_explainer.value.explain_instance(selected_cols, loaded_model.predict_proba, num_features = 10)
    mapping = exp.as_map()[1]

    return str(mapping)        

但是，我在广播中遇到了 pickling 错误。

PicklingError: Can't pickle at 0x7f69fd5680d0>: attribute lookup on lime.discretize failed

有人可以帮忙吗？我们可以使用 dill 之类的东西来代替 spark 中使用的 cloudpickler 吗？

Answer 1

我是 dill 的作者。我同意@Majaha，并将稍微扩展@Majaha 的回答。在@Majaha 的回答的第一个链接中，明确指出 Broadcast 实例是硬连线使用 pickle... 所以建议 dill 到一个字符串，然后 undill 之后是一个很好的。

不幸的是，extend 方法可能对您不起作用。在 Broadcast 类中，源使用 CPickle，dill 无法扩展。如果您查看源代码，它使用 import CPickle as pickle； ... pickle.dumps 用于 python 2，以及 import pickle； ... pickle.dumps 用于 python 3。是否使用了 import pickle； ... pickle.dumps 用于 python 2，以及 import pickle； ... pickle._dumps 对于 python 3，然后 dill 可以通过执行 import dill 来扩展 pickler。例如:

Python 3.6.6 (default, Jun 28 2018, 05:53:46) 
[GCC 4.2.1 Compatible Apple LLVM 9.0.0 (clang-900.0.39.2)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> from pickle import _dumps
>>> import dill
>>> _dumps(lambda x:x)
b'\x80\x03cdill._dill\n_create_function\nq\x00(cdill._dill\n_load_type\nq\x01X\x08\x00\x00\x00CodeTypeq\x02\x85q\x03Rq\x04(K\x01K\x00K\x01K\x01KCC\x04|\x00S\x00q\x05N\x85q\x06)X\x01\x00\x00\x00xq\x07\x85q\x08X\x07\x00\x00\x00<stdin>q\tX\x08\x00\x00\x00<lambda>q\nK\x01C\x00q\x0b))tq\x0cRq\rc__main__\n__dict__\nh\nNN}q\x0etq\x0fRq\x10.'

因此，您可以按照@Majaha 的建议进行操作(并预订对 broadcast 的调用)，或者您可以修补代码以进行我上面概述的替换(在需要的地方，但是呃。 ..)，或者您可以使用 dill 创建自己的派生类来完成这项工作:

>>> from pyspark.broadcast import Broadcast as _Broadcast
>>>
>>> class Broadcast(_Broadcast):
...   def dump(self, value, f):
...     try:
...       import dill
...       dill.dump(value, f, pickle_protocol)
...     ...[INSERT THE REST OF THE DUMP METHOD HERE]...

如果上述方法失败...您仍然可以通过查明序列化失败发生的位置来使其正常工作(dill.detect.trace 可以帮助您解决这个问题)。

如果您要建议 pyspark 使用 dill... 一个可能更好的建议是允许用户动态替换序列化程序。 mpi4py 和其他一些软件包就是这样做的。