Python - PySpark 的 Pickle Spacy答案

【问题标题】：Python - Pickle Spacy for PySparkPython - PySpark 的 Pickle Spacy
【发布时间】：2018-11-25 13:52:33
【问题描述】：

documentation for Spacy 2.0 提到开发人员已经添加了允许 Spacy 被腌制的功能，以便它可以被 PySpark 接口的 Spark Cluster 使用，但是，他们没有提供有关如何执行此操作的说明。

有人可以解释我如何腌制 Spacy 的英语 NE 解析器以在我的 udf 函数中使用吗？

这不起作用：

from pyspark import cloudpickle
nlp = English()
pickled_nlp = cloudpickle.dumps(nlp)

【问题讨论】：

如果你不介意跟进，我对你最终做了什么很感兴趣。
我找到了另一种解决方法来正确序列化 Spacy。相反，我创建了一个 UDF 函数，在其中设置了一个全局变量来存储我的 NLP 对象，然后我使用 try/except 循环来尝试使用 NLP 对象，如果这不起作用，请先初始化它然后使用它.这样，我只需要在每个 worker 中初始化一次，而不是在函数每次运行时初始化一次。
听起来不错！你能用一些代码回答这个 SO 问题吗？谢谢！
我的意思是这并不是我的问题的真正答案。这是一种解决方法，但可以肯定。

标签： python apache-spark pyspark user-defined-functions

【解决方案1】：

不是真正的答案，而是我发现的最佳解决方法：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, ArrayType
import spacy

def get_entities_udf():
    def get_entities(text):
        global nlp
        try:
            doc = nlp(unicode(text))
        except:
            nlp = spacy.load('en')
            doc = nlp(unicode(text))
        return [t.label_ for t in doc.ents]
    res_udf = udf(get_entities, StringType(ArrayType()))
    return res_udf

documents_df = documents_df.withColumn('entities', get_entities_udf()('text'))

【讨论】：

很遗憾我也在用这个。

【解决方案2】：

这满足了我的需求，而且似乎非常快（改编自讨论结束 here）：

# create class to wrap spacy object
class SpacyMagic(object):
    """
    Simple Spacy Magic to minimize loading time.
    >>> SpacyMagic.get("en")
    <spacy.en.English ...
    """
    _spacys = {}

    @classmethod
    def get(cls, lang):
        if lang not in cls._spacys:
            import spacy
            cls._spacys[lang] = spacy.load(lang, disable=['parser', 'tagger', 'ner'])
        return cls._spacys[lang]

# broadcast `nlp` object as `nlp_br`
nlp_br = sc.broadcast( SpacyMagic.get('en_core_web_lg') )

# returns a list of word2vec vectors for each phrase or word `x`
def get_vector(x):
    return nlp_br.value(x).vector.tolist()

get_vector_udf = F.udf( get_vector, T.ArrayType( T.FloatType() ) )

# create new column with word2vec vectors
new_df = df.withColumn( 'w2v_vectors', get_vector_udf( F.col('textColumn') ) )

【讨论】：

有趣。我很好奇 spacy 在处理文档时为每个nlp 对象创建的底层vocab 发生了什么，而广播的nlp 是只读的。