自定义 Python 对象在 Pyspark UDF 中的用法

Question

运行以下一段 PySpark 代码时：

nlp = NLPFunctions()
def parse_ingredients(ingredient_lines):
    parsed_ingredients = nlp.getingredients_bulk(ingredient_lines)[0]
    return list(chain.from_iterable(parsed_ingredients))

udf_parse_ingredients = UserDefinedFunction(parse_ingredients, ArrayType(StringType()))

我收到以下错误： _pickle.PicklingError: Could not serialize object: TypeError: can't pickle _thread.lock objects

我想这是因为 PySpark 无法序列化这个自定义类。但是，如何避免在每次运行 parse_ingredients_line 函数时实例化此昂贵对象的开销呢？

Answer 1

假设你想使用Identity这样定义的类（identity.py）：

class Identity(object):                   
    def __getstate__(self):
        raise NotImplementedError("Not serializable")
    def identity(self, x):
        return x

例如，您可以使用可调用对象（f.py）并将Identity实例存储为类成员：

from identity import Identity
class F(object):                          
    identity = None
    def __call__(self, x):
        if not F.identity:
            F.identity = Identity()
        return F.identity.identity(x)

并按如下所示使用它们：

from pyspark.sql.functions import udf
import f
sc.addPyFile("identity.py")
sc.addPyFile("f.py")
f_ = udf(f.F())
spark.range(3).select(f_("id")).show()

+-----+
|F(id)|
+-----+
|    0|
|    1|
|    2|
+-----+

或独立功能和闭合：

from pyspark.sql.functions import udf
import identity
sc.addPyFile("identity.py")
def f(): 
    dict_ = {}                 
    @udf()              
    def f_(x):                 
        if "identity" not in dict_:
            dict_["identity"] = identity.Identity()
        return dict_["identity"].identity(x)
    return f_

spark.range(3).select(f()("id")).show()

+------+
|f_(id)|
+------+
|     0|
|     1|
|     2|
+------+

Answer 2

我通过使 NLPFunctions 类的所有依赖项可序列化来基于 （https://github.com/scikit-learn/scikit-learn/issues/6975） 解决了它。

Answer 3

编辑：这个答案是错误的。对象在广播时仍会序列化，然后反序列化，因此不会避免序列化。（正确使用大型广播变量的提示？

---

尝试使用广播变量。

sc = SparkContext()
nlp_broadcast = sc.broadcast(nlp) # Stores nlp in de-serialized format.
def parse_ingredients(ingredient_lines):
    parsed_ingredients = nlp_broadcast.value.getingredients_bulk(ingredient_lines)[0]
    return list(chain.from_iterable(parsed_ingredients))