python - pyspark 中未定义的函数 UDF？

Question

我有一个在 Dataframe 中调用的 UDF，但我得到了未定义的 udf。

global ac
ac = sc.accumulator(0)

def incrementAC():
  ac.add(1)
  return str(ac.value)

df = sc.parallelize([('Java',90),('Scala',95),('Spark',92)]).toDF(["language","rank"])

df.withColumn("lang_and_rank", expr("concat(language,'blah')")).show()

+--------+----+-------------+
|language|rank|lang_and_rank|
+--------+----+-------------+
|    Java|  90|     Javablah|
|   Scala|  95|    Scalablah|
|   Spark|  92|    Sparkblah|
+--------+----+-------------+

myudf = udf(incrementAC,StringType())
df.withColumn("lang_and_rank", expr("concat(language,myudf())")).show()

.utils.AnalysisException: u'undefined function myudf;'

Answer 1

必须注册与 expr 一起使用的函数:

spark.udf.register("incrementAC", incrementAC)

此外，从转换中使用的累加器也不可靠。