ML 函数作为 pyspark UDF

Question

我对 pyspark 和 python 有点陌生。我正在尝试将 ML 函数作为 pyspark UDF 运行。

这是一个例子：

from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import StringType

df = spark.createDataFrame(['Bob has a dog. He loves him'], StringType())

def parse(text):
    import spacy
    import neuralcoref
    nlp = spacy.load('en_core_web_sm')
    # Let's try before using the conversion dictionary:
    neuralcoref.add_to_pipe(nlp)
    doc = nlp(text)
    return doc._.coref_resolved

 pd_udf = pandas_udf(parse, returnType=StringType())

 df.select(pd_udf(col("value"))).show()

收到此错误：

org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/home/user/tools/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 377, in main
    process()
  File "/home/user/tools/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 372, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/home/user/tools/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/serializers.py", line 286, in dump_stream
    for series in iterator:
  File "<string>", line 1, in <lambda>
  File "/home/user/tools/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 101, in <lambda>
    return lambda *a: (verify_result_length(*a), arrow_return_type)
  File "/home/user/tools/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 92, in verify_result_length
    result = f(*a)
  File "/home/user/tools/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/util.py", line 99, in wrapper
    return f(*args, **kwargs)
  File "<stdin>", line 7, in parse
  File "/home/user/anaconda3/lib/python3.7/site-packages/spacy/language.py", line 377, in __call__
    doc = self.make_doc(text)
  File "/home/user/anaconda3/lib/python3.7/site-packages/spacy/language.py", line 401, in make_doc
    return self.tokenizer(text)
TypeError: Argument 'string' has incorrect type (expected str, got Series)

是否可以在 Pyspark 上运行此代码？

Answer 1

如果您更熟悉 Pandas，另一种方法是使用 Pandas UDF -

from pyspark import SparkContext
from pyspark.sql import SQLContext
from functools import reduce
import pyspark.sql.functions as F

from pyspark.sql.types import StringType,StructType,StructField,FloatType,ArrayType,IntegerType,DateType

from functools import partial

from pyspark.sql.functions import lit,array,pandas_udf,PandasUDFType

import pandas as pd

import spacy
import neuralcoref


#### Broadcast the load and makes it available across the worker nodes

nlp = sc.broadcast(spacy.load('en_core_web_sm'))


def udf_parse(text,input_col='value'):

    neuralcoref.add_to_pipe(nlp)
    
    doc = nlp(text.loc[:,input_col])
    
    text['parsed_text'] = doc._.coref_resolved
    
    return text
    
sc = SparkContext.getOrCreate()
sql = SQLContext(sc)

sparkDF = sql.createDataFrame(['Bob has a dog. He loves him'], StringType())

schema = StructType([
           StructField('value', StringType(), True),
           StructField('parsed_value', StringType(), True)
 ])

partial_func = partial(udf_parse,input_col='value')


sparkDF_agg = sparkDF.groupby().applyInPandas(partial_func,schema)

Answer 2

您好，我在设置 spacy 和 neurocoref 时遇到了一大堆问题，所以我用一个反转字符串的随机函数替换了 nlp 函数。

但基本上这里发生的情况是，当您将 col("value") 传递给 pd_udf 时，它是一个 pd.Series 类型。因此，在您的解析函数中，您应该将其定义为接受该类型，如下所示：

def nlp(text):
    return text[::-1]

@pandas_udf("string")
def parse(text: pd.Series) -> pd.Series:
    text = text.apply(nlp)
    return text

pd_udf = pandas_udf(parse, returnType=StringType())

df = spark.createDataFrame([("Bob has a dog. He loves him",),("dog jumps over the fox",)], ("my_text",))
df.select(parse("my_text")).show()

这给出了以下结果：

+--------------------+
|      parse(my_text)|
+--------------------+
|mih sevol eH .god...|
|xof eht revo spmu...|
+--------------------+