Как добавить UDF в конвейер pyspark?

Question

У меня есть следующий код, который в основном выполняет конвейер разработки функций:

token_q1=Tokenizer(inputCol='question1',outputCol='question1_tokens') 
token_q2=Tokenizer(inputCol='question2',outputCol='question2_tokens')  

remover_q1=StopWordsRemover(inputCol='question1_tokens',outputCol='question1_tokens_filtered')
remover_q2=StopWordsRemover(inputCol='question2_tokens',outputCol='question2_tokens_filtered')

q1w2model = Word2Vec(inputCol='question1_tokens_filtered',outputCol='q1_vectors')
q1w2model.setSeed(1)

q2w2model = Word2Vec(inputCol='question2_tokens_filtered',outputCol='q2_vectors')
q2w2model.setSeed(1)

pipeline=Pipeline(stages[token_q1,token_q2,remover_q1,remover_q2,q1w2model,q2w2model])
model=pipeline.fit(train)
result=model.transform(train)
result.show()

Я хочу добавить следующий UDF к этому конвейеру:

charcount_q1 = F.udf(lambda row : sum([len(char) for char in row]),IntegerType())

Когда я это делаю, я получаю ошибку Java. Может ли кто-нибудь указать мне правильное направление?

Однако я добавил этот столбец, используя следующий код, который в основном работает:

charCountq1=train.withColumn("charcountq1", charcount_q1("question1"))

Но я хочу добавить это в пиплин, а не делать это так

Answer 1

Если вы хотите использовать udf в Pipeline, вам понадобится одно из следующих:

• Создание собственного трансформера в PySpark ML
• Использование SQLTransformer

Первый достаточно многословен для такого простого варианта использования, поэтому я рекомендую второй вариант:

from pyspark.sql.functions import udf
from pyspark.ml import Pipeline
from pyspark.ml.feature import SQLTransformer

charcount_q1 = spark.udf.register(
    "charcount_q1",
    lambda row : sum(len(char) for char in row),
    "integer"
)

df = spark.createDataFrame(
    [(1, ["spark", "java", "python"])],
    ("id", "question1"))

pipeline = Pipeline(stages = [SQLTransformer(
    statement = "SELECT *, charcount_q1(question1) charcountq1 FROM __THIS__"
)])

pipeline.fit(df).transform(df).show()
# +---+--------------------+-----------+
# | id|           question1|charcountq1|
# +---+--------------------+-----------+
# |  1|[spark, java, pyt...|         15|
# +---+--------------------+-----------+