PROSAGA码农传奇-特征工程-如何在pyspark管道中添加UDF？

如何在pyspark管道中添加UDF？

作者: 诸葛神侯
发布时间: 2025-01-20 04:01:35 (1月前)
转自：

<div class =“post-text”itemprop =“text”>
  <P>
    如果你想使用
     <code>
      udf
    </code>
     在
     <code>
      Pipeline
    </code>
     你需要以下之一：
  </p>
  <UL>
    <LI>
      <a href="https://stackoverflow.com/q/32331848/9613318">
        在PySpark ML中创建自定义Transformer
      </A>
    </LI>
    <LI>
      使用
      <a href="http://spark.apache.org/docs/latest/ml-features.html#sqltransformer" rel="nofollow noreferrer">
         <code>
          SQLTransformer
        </code>
      </A>
    </LI>
  </UL>
  <P>
    第一个对于这样一个简单的用例非常冗长，所以我推荐第二个选项：
  </p>
   <pre class="lang-py prettyprint-override">
    <code>
      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|
# +---+--------------------+-----------+

</code>
  </pre>
</DIV>